// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <kmock/kmock-matchers.h>
#include <ktest/ktest.h>

#include <algorithm>
#include <cmath>
#include <cstdint>
#include <cstring>
#include <future>
#include <limits>
#include <memory>
#include <numeric>
#include <string>
#include <type_traits>
#include <utility>
#include <vector>

#include <nebula/array/array_base.h>
#include <nebula/array/array_binary.h>
#include <nebula/array/array_decimal.h>
#include <nebula/array/array_primitive.h>
#include <nebula/array/builder_binary.h>
#include <nebula/array/builder_decimal.h>
#include <nebula/array/builder_dict.h>
#include <nebula/array/builder_run_end.h>
#include <nebula/array/builder_time.h>
#include <nebula/array/data.h>
#include <nebula/array/util.h>
#include <nebula/core/buffer.h>
#include <nebula/core/buffer_builder.h>
#include <nebula/core/compare.h>

#include <nebula/types/scalar.h>
#include <turbo/utility/status.h>
#include <nebula/testing/builder.h>
#include <nebula/testing/extension_type.h>
#include <ktest/ktest.h>
#include <nebula/testing/ktest_util.h>
#include <nebula/testing/random.h>
#include <nebula/testing/util.h>
#include <nebula/types/type.h>
#include <nebula/types/type_traits.h>
#include <nebula/bits/bit_util.h>
#include <nebula/bits/bitmap_builders.h>
#include <turbo/base/checked_cast.h>
#include <nebula/numeric/decimal.h>
#include <nebula/util/key_value_metadata.h>
#include <turbo/base/macros.h>
#include <nebula/util/range.h>
#include <nebula/core/visit_data_inline.h>

// This file is compiled together with array-*-test.cc into a single
// executable array-test.

namespace nebula {

    class TestArray : public ::testing::Test {
    public:
        void SetUp() { pool_ = default_memory_pool(); }

    protected:
        MemoryPool *pool_;
    };

    void check_dictionary_null_count(const std::shared_ptr<DataType> &dict_type,
                                  const std::string &input_dictionary_json,
                                  const std::string &input_index_json,
                                  const int64_t &expected_null_count,
                                  const int64_t &expected_logical_null_count,
                                  bool expected_may_have_nulls,
                                  bool expected_may_have_logical_nulls) {
        std::shared_ptr<nebula::Array> arr =
                assert_dict_array_from_json(dict_type, input_index_json, input_dictionary_json);

        ASSERT_EQ(arr->null_count(), expected_null_count);
        ASSERT_EQ(arr->compute_logical_null_count(), expected_logical_null_count);
        ASSERT_EQ(arr->data()->may_have_nulls(), expected_may_have_nulls);
        ASSERT_EQ(arr->data()->may_have_logical_nulls(), expected_may_have_logical_nulls);
    }

    TEST_F(TestArray, TestNullCount) {
        // These are placeholders
        auto data = std::make_shared<Buffer>(nullptr, 0);
        auto null_bitmap = std::make_shared<Buffer>(nullptr, 0);

        std::shared_ptr<Int32Array> arr(new Int32Array(100, data, null_bitmap, 10));
        ASSERT_EQ(10, arr->compute_logical_null_count());
        ASSERT_EQ(10, arr->null_count());
        ASSERT_TRUE(arr->data()->may_have_nulls());
        ASSERT_TRUE(arr->data()->may_have_logical_nulls());

        std::shared_ptr<Int32Array> arr_no_nulls(new Int32Array(100, data));
        ASSERT_EQ(0, arr_no_nulls->compute_logical_null_count());
        ASSERT_EQ(0, arr_no_nulls->null_count());
        ASSERT_FALSE(arr_no_nulls->data()->may_have_nulls());
        ASSERT_FALSE(arr_no_nulls->data()->may_have_logical_nulls());

        std::shared_ptr<Int32Array> arr_default_null_count(
                new Int32Array(100, data, null_bitmap));
        ASSERT_EQ(kUnknownNullCount, arr_default_null_count->data()->null_count);
        ASSERT_TRUE(arr_default_null_count->data()->may_have_nulls());
        ASSERT_TRUE(arr_default_null_count->data()->may_have_logical_nulls());

        RunEndEncodedBuilder ree_builder(pool_, std::make_shared<Int32Builder>(pool_),
                                         std::make_shared<Int32Builder>(pool_),
                                         run_end_encoded(int32(), int32()));
        ASSERT_OK(ree_builder.append_scalar(*MakeScalar<int32_t>(2), 2));
        ASSERT_OK(ree_builder.append_null());
        ASSERT_OK(ree_builder.append_scalar(*MakeScalar<int32_t>(4), 3));
        ASSERT_OK(ree_builder.append_nulls(2));
        ASSERT_OK(ree_builder.append_scalar(*MakeScalar<int32_t>(8), 5));
        ASSERT_OK(ree_builder.append_nulls(7));
        ASSERT_OK_AND_ASSIGN(auto ree, ree_builder.finish());

        ASSERT_EQ(0, ree->null_count());
        ASSERT_EQ(10, ree->compute_logical_null_count());
        ASSERT_FALSE(ree->data()->may_have_nulls());
        ASSERT_TRUE(ree->data()->may_have_logical_nulls());

        ASSERT_OK(ree_builder.append_scalar(*MakeScalar<int32_t>(2), 2));
        ASSERT_OK(ree_builder.append_scalar(*MakeScalar<int32_t>(4), 3));
        ASSERT_OK(ree_builder.append_scalar(*MakeScalar<int32_t>(8), 5));
        ASSERT_OK_AND_ASSIGN(auto ree_no_nulls, ree_builder.finish());
        ASSERT_EQ(0, ree_no_nulls->null_count());
        ASSERT_EQ(0, ree_no_nulls->compute_logical_null_count());
        ASSERT_FALSE(ree_no_nulls->data()->may_have_nulls());
        ASSERT_FALSE(ree_no_nulls->data()->may_have_logical_nulls());

        // Dictionary type
        std::shared_ptr<nebula::DataType> type;
        std::shared_ptr<nebula::DataType> dict_type;

        for (const auto &index_type: all_dictionary_index_types()) {
            NEBULA_SCOPED_TRACE("index_type = ", index_type->to_string());

            type = boolean();
            dict_type = dictionary(index_type, type);
            // no null value
            check_dictionary_null_count(dict_type, "[]", "[]", 0, 0, false, false);
            check_dictionary_null_count(dict_type, "[true, false]", "[0, 1, 0]", 0, 0, false, false);

            // only indices contain null value
            check_dictionary_null_count(dict_type, "[true, false]", "[null, 0, 1]", 1, 1, true,
                                     true);
            check_dictionary_null_count(dict_type, "[true, false]", "[null, null]", 2, 2, true,
                                     true);

            // only dictionary contains null value
            check_dictionary_null_count(dict_type, "[null, true]", "[]", 0, 0, false, true);
            check_dictionary_null_count(dict_type, "[null, true, false]", "[0, 1, 0]", 0, 2, false,
                                     true);

            // both indices and dictionary contain null value
            check_dictionary_null_count(dict_type, "[null, true, false]", "[0, 1, 0, null]", 1, 3,
                                     true, true);
            check_dictionary_null_count(dict_type, "[null, true, null, false]", "[null, 1, 0, 2, 3]",
                                     1, 3, true, true);
        }
    }

    TEST_F(TestArray, TestSlicePreservesAllNullCount) {
        // These are placeholders
        auto data = std::make_shared<Buffer>(nullptr, 0);
        auto null_bitmap = std::make_shared<Buffer>(nullptr, 0);

        Int32Array arr(/*length=*/100, data, null_bitmap,
                /*null_count*/ 100);
        EXPECT_EQ(arr.slice(1, 99)->data()->null_count, arr.slice(1, 99)->length());

        // Dictionary type
        std::shared_ptr<nebula::DataType> dict_type = dictionary(int64(), boolean());
        std::shared_ptr<nebula::Array> dict_arr =
                assert_dict_array_from_json(dict_type, /*indices=*/"[null, 0, 0, 0, 0, 0, 1, 2, 0, 0]",
                        /*dictionary=*/"[null, true, false]");
        ASSERT_EQ(dict_arr->null_count(), 1);
        ASSERT_EQ(dict_arr->compute_logical_null_count(), 8);
        ASSERT_EQ(dict_arr->slice(2, 8)->null_count(), 0);
        ASSERT_EQ(dict_arr->slice(2, 8)->compute_logical_null_count(), 6);
    }

    TEST_F(TestArray, TestLength) {
        // Placeholder buffer
        auto data = std::make_shared<Buffer>(nullptr, 0);

        std::unique_ptr<Int32Array> arr(new Int32Array(100, data));
        ASSERT_EQ(arr->length(), 100);
    }

    TEST_F(TestArray, TestNullToString) {
        // Invalid NULL buffer
        auto data = std::make_shared<Buffer>(nullptr, 400);

        std::unique_ptr<Int32Array> arr(new Int32Array(100, data));
        ASSERT_EQ(arr->to_string(),
                  "<Invalid array: Missing values buffer in non-empty fixed-width array>");
    }

    TEST_F(TestArray, TestSliceSafe) {
        std::vector<int32_t> original_data{1, 2, 3, 4, 5, 6, 7};
        auto arr = std::make_shared<Int32Array>(7, Buffer::Wrap(original_data));

        auto check_data = [](const Array &arr, const std::vector<int32_t> &expected) {
            ASSERT_EQ(arr.length(), static_cast<int64_t>(expected.size()));
            const int32_t *data = arr.data()->get_values<int32_t>(1);
            for (int64_t i = 0; i < arr.length(); ++i) {
                ASSERT_EQ(data[i], expected[i]);
            }
        };

        check_data(*arr, {1, 2, 3, 4, 5, 6, 7});

        ASSERT_OK_AND_ASSIGN(auto sliced, arr->slice_safe(0, 0));
        check_data(*sliced, {});

        ASSERT_OK_AND_ASSIGN(sliced, arr->slice_safe(0, 7));
        check_data(*sliced, original_data);

        ASSERT_OK_AND_ASSIGN(sliced, arr->slice_safe(3, 4));
        check_data(*sliced, {4, 5, 6, 7});

        ASSERT_OK_AND_ASSIGN(sliced, arr->slice_safe(0, 7));
        check_data(*sliced, {1, 2, 3, 4, 5, 6, 7});

        ASSERT_OK_AND_ASSIGN(sliced, arr->slice_safe(7, 0));
        check_data(*sliced, {});

        ASSERT_RAISES(turbo::StatusCode::kOutOfRange, arr->slice_safe(8, 0));
        ASSERT_RAISES(turbo::StatusCode::kOutOfRange, arr->slice_safe(0, 8));
        ASSERT_RAISES(turbo::StatusCode::kOutOfRange, arr->slice_safe(-1, 0));
        ASSERT_RAISES(turbo::StatusCode::kOutOfRange, arr->slice_safe(0, -1));
        ASSERT_RAISES(turbo::StatusCode::kOutOfRange, arr->slice_safe(6, 2));
        ASSERT_RAISES(turbo::StatusCode::kOutOfRange, arr->slice_safe(6, std::numeric_limits<int64_t>::max() - 5));

        ASSERT_OK_AND_ASSIGN(sliced, arr->slice_safe(0));
        check_data(*sliced, original_data);

        ASSERT_OK_AND_ASSIGN(sliced, arr->slice_safe(3));
        check_data(*sliced, {4, 5, 6, 7});

        ASSERT_OK_AND_ASSIGN(sliced, arr->slice_safe(7));
        check_data(*sliced, {});

        ASSERT_RAISES(turbo::StatusCode::kOutOfRange, arr->slice_safe(8));
        ASSERT_RAISES(turbo::StatusCode::kOutOfRange, arr->slice_safe(-1));
    }

    turbo::Status MakeArrayFromValidBytes(const std::vector<uint8_t> &v, MemoryPool *pool,
                                          std::shared_ptr<Array> *out) {
        int64_t null_count = v.size() - std::accumulate(v.begin(), v.end(), 0);

        TURBO_MOVE_OR_RAISE(auto null_buf, internal::BytesToBits(v));

        TypedBufferBuilder<int32_t> value_builder(pool);
        for (size_t i = 0; i < v.size(); ++i) {
            TURBO_RETURN_NOT_OK(value_builder.append(0));
        }

        std::shared_ptr<Buffer> values;
        TURBO_RETURN_NOT_OK(value_builder.finish(&values));
        *out = std::make_shared<Int32Array>(v.size(), values, null_buf, null_count);
        return turbo::OkStatus();
    }

    TEST_F(TestArray, TestEquality) {
        std::shared_ptr<Array> array, equal_array, unequal_array;

        ASSERT_OK(MakeArrayFromValidBytes({1, 0, 1, 1, 0, 1, 0, 0}, pool_, &array));
        ASSERT_OK(MakeArrayFromValidBytes({1, 0, 1, 1, 0, 1, 0, 0}, pool_, &equal_array));
        ASSERT_OK(MakeArrayFromValidBytes({1, 1, 1, 1, 0, 1, 0, 0}, pool_, &unequal_array));

        EXPECT_TRUE(array->equals(array));
        EXPECT_TRUE(array->equals(equal_array));
        EXPECT_TRUE(equal_array->equals(array));
        EXPECT_FALSE(equal_array->equals(unequal_array));
        EXPECT_FALSE(unequal_array->equals(equal_array));
        EXPECT_TRUE(array->range_equals(4, 8, 4, unequal_array));
        EXPECT_FALSE(array->range_equals(0, 4, 0, unequal_array));
        EXPECT_FALSE(array->range_equals(0, 8, 0, unequal_array));
        EXPECT_FALSE(array->range_equals(1, 2, 1, unequal_array));

        auto timestamp_ns_array = std::make_shared<NumericArray<TimestampType>>(
                timestamp(TimeUnit::NANO), array->length(), array->data()->buffers[1],
                array->data()->buffers[0], array->null_count());
        auto timestamp_us_array = std::make_shared<NumericArray<TimestampType>>(
                timestamp(TimeUnit::MICRO), array->length(), array->data()->buffers[1],
                array->data()->buffers[0], array->null_count());
        ASSERT_FALSE(array->equals(timestamp_ns_array));
        // ARROW-2567: Ensure that not only the type id but also the type equality
        // itself is checked.
        ASSERT_FALSE(timestamp_us_array->equals(timestamp_ns_array));
        ASSERT_TRUE(timestamp_us_array->range_equals(0, 1, 0, timestamp_us_array));
        ASSERT_FALSE(timestamp_us_array->range_equals(0, 1, 0, timestamp_ns_array));
    }

    TEST_F(TestArray, TestNullArrayEquality) {
        auto array_1 = std::make_shared<NullArray>(10);
        auto array_2 = std::make_shared<NullArray>(10);
        auto array_3 = std::make_shared<NullArray>(20);

        EXPECT_TRUE(array_1->equals(array_1));
        EXPECT_TRUE(array_1->equals(array_2));
        EXPECT_FALSE(array_1->equals(array_3));
    }

    TEST_F(TestArray, SliceRecomputeNullCount) {
        std::vector<uint8_t> valid_bytes = {1, 0, 1, 1, 0, 1, 0, 0, 0};

        std::shared_ptr<Array> array;
        ASSERT_OK(MakeArrayFromValidBytes(valid_bytes, pool_, &array));

        ASSERT_EQ(5, array->null_count());

        auto slice = array->slice(1, 4);
        ASSERT_EQ(2, slice->null_count());

        slice = array->slice(4);
        ASSERT_EQ(4, slice->null_count());

        auto slice2 = slice->slice(0);
        ASSERT_EQ(4, slice2->null_count());

        slice = array->slice(0);
        ASSERT_EQ(5, slice->null_count());

        // No bitmap, compute 0
        const int kBufferSize = 64;
        ASSERT_OK_AND_ASSIGN(auto data, allocate_buffer(kBufferSize, pool_));
        memset(data->mutable_data(), 0, kBufferSize);

        auto arr = std::make_shared<Int32Array>(16, std::move(data), nullptr, -1);
        ASSERT_EQ(0, arr->null_count());
    }

    TEST_F(TestArray, NullArraySliceNullCount) {
        auto null_arr = std::make_shared<NullArray>(10);
        auto null_arr_sliced = null_arr->slice(3, 6);

        // The internal null count is 6, does not require recomputation
        ASSERT_EQ(6, null_arr_sliced->data()->null_count);

        ASSERT_EQ(6, null_arr_sliced->null_count());
    }

    TEST_F(TestArray, TestIsNullIsValid) {
        // clang-format off
        std::vector<uint8_t> null_bitmap = {1, 0, 1, 1, 0, 1, 0, 0,
                                            1, 0, 1, 1, 0, 1, 0, 0,
                                            1, 0, 1, 1, 0, 1, 0, 0,
                                            1, 0, 1, 1, 0, 1, 0, 0,
                                            1, 0, 0, 1};
        // clang-format on
        int64_t null_count = 0;
        for (uint8_t x: null_bitmap) {
            if (x == 0) {
                ++null_count;
            }
        }

        ASSERT_OK_AND_ASSIGN(auto null_buf, internal::BytesToBits(null_bitmap));

        std::unique_ptr<Array> arr;
        arr.reset(new Int32Array(null_bitmap.size(), nullptr, null_buf, null_count));

        ASSERT_EQ(null_count, arr->null_count());
        ASSERT_EQ(5, null_buf->size());

        ASSERT_TRUE(arr->null_bitmap()->equals(*null_buf.get()));

        for (size_t i = 0; i < null_bitmap.size(); ++i) {
            EXPECT_EQ(null_bitmap[i] != 0, !arr->is_null(i)) << i;
            EXPECT_EQ(null_bitmap[i] != 0, arr->is_valid(i)) << i;
        }
    }

    TEST_F(TestArray, TestIsNullIsValidNoNulls) {
        const int64_t size = 10;

        std::unique_ptr<Array> arr;
        arr.reset(new Int32Array(size, nullptr, nullptr, 0));

        for (size_t i = 0; i < size; ++i) {
            EXPECT_TRUE(arr->is_valid(i));
            EXPECT_FALSE(arr->is_null(i));
        }
    }

    TEST_F(TestArray, BuildLargeInMemoryArray) {
#ifdef NDEBUG
        const int64_t length = static_cast<int64_t>(std::numeric_limits<int32_t>::max()) + 1;
#elif !defined(NEBULA_VALGRIND)
        // use a smaller size since the insert function isn't optimized properly on debug and
        // the test takes a long time to complete
        const int64_t length = 2 << 24;
#else
                                                                                                                                // use an even smaller size with valgrind
  const int64_t length = 2 << 20;
#endif

        BooleanBuilder builder;
        std::vector<bool> zeros(length);
        ASSERT_OK(builder.append_values(zeros));

        std::shared_ptr<Array> result;
        FinishAndCheckPadding(&builder, &result);

        ASSERT_EQ(length, result->length());
    }

    static std::vector<std::shared_ptr<DataType>> TestArrayUtilitiesAgainstTheseTypes() {
        FieldVector union_fields1({field("a", utf8()), field("b", int32())});
        FieldVector union_fields2({field("a", null()), field("b", list(large_utf8()))});
        std::vector<int8_t> union_type_codes{7, 42};

        return {
                null(),
                boolean(),
                int8(),
                uint16(),
                int32(),
                uint64(),
                float64(),
                binary(),
                large_binary(),
                binary_view(),
                fixed_size_binary(3),
                decimal(16, 4),
                utf8(),
                large_utf8(),
                utf8_view(),
                list(utf8()),
                list(int64()),  // NOTE: Regression case for ARROW-9071/MakeArrayOfNull
                list(large_utf8()),
                list(list(int64())),
                list(list(large_utf8())),
                large_list(utf8()),
                large_list(large_utf8()),
                large_list(list(large_utf8())),
                fixed_size_list(utf8(), 3),
                fixed_size_list(int64(), 4),
                list_view(utf8()),
                large_list_view(utf8()),
                dictionary(int32(), utf8()),
                STRUCT({field("a", utf8()), field("b", int32())}),
                sparse_union(union_fields1, union_type_codes),
                sparse_union(union_fields2, union_type_codes),
                dense_union(union_fields1, union_type_codes),
                dense_union(union_fields2, union_type_codes),
                smallint(),             // extension type
                list_extension_type(),  // nested extension type
                run_end_encoded(int16(), utf8()),
        };
    }

    TEST_F(TestArray, TestMakeArrayOfNull) {
        for (int64_t length: {0, 1, 16, 133}) {
            for (auto type: TestArrayUtilitiesAgainstTheseTypes()) {
                NEBULA_SCOPED_TRACE("type = ", type->to_string());
                ASSERT_OK_AND_ASSIGN(auto array, MakeArrayOfNull(type, length));
                ASSERT_EQ(array->type(), type);
                ASSERT_OK(array->validate_full());
                ASSERT_EQ(array->length(), length);
                ASSERT_EQ(array->device_type(), DeviceAllocationType::kCPU);
                if (is_union(type->id())) {
                    ASSERT_EQ(array->null_count(), 0);
                    ASSERT_EQ(array->compute_logical_null_count(), length);
                    const auto &union_array = turbo::checked_cast<const UnionArray &>(*array);
                    for (int i = 0; i < union_array.num_fields(); ++i) {
                        ASSERT_EQ(union_array.field(i)->null_count(), union_array.field(i)->length());
                    }
                } else if (type->id() == Type::RUN_END_ENCODED) {
                    ASSERT_EQ(array->null_count(), 0);
                    ASSERT_EQ(array->compute_logical_null_count(), length);
                    const auto &ree_array = turbo::checked_cast<const RunEndEncodedArray &>(*array);
                    ASSERT_EQ(ree_array.values()->null_count(), ree_array.values()->length());
                } else {
                    ASSERT_EQ(array->null_count(), length);
                    ASSERT_EQ(array->compute_logical_null_count(), length);
                }
                for (int64_t i = 0; i < length; ++i) {
                    ASSERT_TRUE(array->is_null(i));
                    ASSERT_FALSE(array->is_valid(i));
                }
            }
        }
    }

    TEST_F(TestArray, TestMakeArrayOfNullUnion) {
        // Unions need special checking -- the top level null count is 0 (per
        // ARROW-9222) so we check the first child to make sure is contains all nulls
        // and check that the type_ids all point to the first child
        const int64_t union_length = 10;
        auto s_union_ty = sparse_union({field("a", utf8()), field("b", int32())}, {0, 1});
        ASSERT_OK_AND_ASSIGN(auto s_union_nulls, MakeArrayOfNull(s_union_ty, union_length));
        ASSERT_OK(s_union_nulls->validate_full());
        ASSERT_EQ(s_union_nulls->null_count(), 0);
        {
            const auto &typed_union = turbo::checked_cast<const SparseUnionArray &>(*s_union_nulls);
            ASSERT_EQ(typed_union.field(0)->null_count(), union_length);

            // Check type codes are all 0
            for (int i = 0; i < union_length; ++i) {
                ASSERT_EQ(typed_union.raw_type_codes()[i], 0);
            }
        }

        s_union_ty = sparse_union({field("a", utf8()), field("b", int32())}, {2, 7});
        ASSERT_OK_AND_ASSIGN(s_union_nulls, MakeArrayOfNull(s_union_ty, union_length));
        ASSERT_OK(s_union_nulls->validate_full());
        ASSERT_EQ(s_union_nulls->null_count(), 0);
        {
            const auto &typed_union = turbo::checked_cast<const SparseUnionArray &>(*s_union_nulls);
            ASSERT_EQ(typed_union.field(0)->null_count(), union_length);

            // Check type codes are all 2
            for (int i = 0; i < union_length; ++i) {
                ASSERT_EQ(typed_union.raw_type_codes()[i], 2);
            }
        }

        auto d_union_ty = dense_union({field("a", utf8()), field("b", int32())}, {0, 1});
        ASSERT_OK_AND_ASSIGN(auto d_union_nulls, MakeArrayOfNull(d_union_ty, union_length));
        ASSERT_OK(d_union_nulls->validate_full());
        ASSERT_EQ(d_union_nulls->null_count(), 0);
        {
            const auto &typed_union = turbo::checked_cast<const DenseUnionArray &>(*d_union_nulls);

            // Child field has length 1 which is a null element
            ASSERT_EQ(typed_union.field(0)->length(), 1);
            ASSERT_EQ(typed_union.field(0)->null_count(), 1);

            // Check type codes are all 0 and the offsets point to the first element of
            // the first child
            for (int i = 0; i < union_length; ++i) {
                ASSERT_EQ(typed_union.raw_type_codes()[i], 0);
                ASSERT_EQ(typed_union.raw_value_offsets()[i], 0);
            }
        }
    }

    TEST_F(TestArray, TestValidateNullCount) {
        Int32Builder builder(pool_);
        ASSERT_OK(builder.append(5));
        ASSERT_OK(builder.append(42));
        ASSERT_OK(builder.append_null());
        ASSERT_OK_AND_ASSIGN(auto array, builder.finish());

        ArrayData *data = array->data().get();
        data->null_count = kUnknownNullCount;
        ASSERT_OK(array->validate_full());
        data->null_count = 1;
        ASSERT_OK(array->validate_full());

        // null_count out of bounds
        data->null_count = -2;
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, array->validate());
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, array->validate_full());
        data->null_count = 4;
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, array->validate());
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, array->validate_full());

        // null_count inconsistent with data
        for (const int64_t null_count: {0, 2, 3}) {
            data->null_count = null_count;
            ASSERT_OK(array->validate());
            ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, array->validate_full());
        }
    }

    void AssertAppendScalar(MemoryPool *pool, const std::shared_ptr<Scalar> &scalar) {
        std::unique_ptr<nebula::ArrayBuilder> builder;
        auto null_scalar = MakeNullScalar(scalar->type);
        ASSERT_OK(MakeBuilderExactIndex(pool, scalar->type, &builder));
        ASSERT_OK(builder->append_scalar(*scalar));                 // [0] = scalar
        ASSERT_OK(builder->append_scalar(*scalar));                 // [1] = scalar
        ASSERT_OK(builder->append_scalar(*null_scalar));            // [2] = NULL
        ASSERT_OK(builder->append_scalars({scalar, null_scalar}));  // [3, 4] = {scalar, NULL}
        ASSERT_OK(
                builder->append_scalar(*scalar, /*n_repeats=*/2));  // [5, 6] = {scalar, scalar}
        ASSERT_OK(
                builder->append_scalar(*null_scalar, /*n_repeats=*/2));  // [7, 8] = {NULL, NULL}

        std::shared_ptr<Array> out;
        FinishAndCheckPadding(builder.get(), &out);
        ASSERT_OK(out->validate_full());
        AssertTypeEqual(scalar->type, out->type());
        ASSERT_EQ(out->length(), 9);

        auto out_type_id = out->type()->id();
        const bool can_check_nulls = internal::may_have_validity_bitmap(out_type_id);
        // For a dictionary builder, the output dictionary won't necessarily be the same
        const bool can_check_values = !is_dictionary(out_type_id);

        if (can_check_nulls) {
            ASSERT_EQ(out->null_count(), 4);
        } else {
            ASSERT_EQ(out->null_count(), 0);
        }
        if (scalar->is_valid) {
            ASSERT_EQ(out->compute_logical_null_count(), 4);
        } else {
            ASSERT_EQ(out->compute_logical_null_count(), 9);
        }

        for (const auto index: {0, 1, 3, 5, 6}) {
            ASSERT_NE(out->is_null(index), scalar->is_valid);
            ASSERT_OK_AND_ASSIGN(auto scalar_i, out->get_scalar(index));
            ASSERT_OK(scalar_i->validate_full());
            if (can_check_values) AssertScalarsEqual(*scalar, *scalar_i, /*verbose=*/true);
        }
        for (const auto index: {2, 4, 7, 8}) {
            ASSERT_TRUE(out->is_null(index));
            ASSERT_OK_AND_ASSIGN(auto scalar_i, out->get_scalar(index));
            ASSERT_OK(scalar_i->validate_full());
            AssertScalarsEqual(*null_scalar, *scalar_i, /*verbose=*/true);
        }
    }

    static ScalarVector GetScalars() {
        auto hello = Buffer::from_string("hello");
        DayTimeIntervalType::DayMilliseconds daytime{1, 100};
        MonthDayNanoIntervalType::MonthDayNanos month_day_nanos{5, 4, 100};

        FieldVector union_fields{field("string", utf8()), field("number", int32()),
                                 field("other_number", int32())};
        std::vector<int8_t> union_type_codes{5, 6, 42};

        const auto sparse_union_ty = ::nebula::sparse_union(union_fields, union_type_codes);
        const auto dense_union_ty = ::nebula::dense_union(union_fields, union_type_codes);

        return {
                std::make_shared<BooleanScalar>(false),
                std::make_shared<Int8Scalar>(3),
                std::make_shared<UInt16Scalar>(3),
                std::make_shared<Int32Scalar>(3),
                std::make_shared<UInt64Scalar>(3),
                std::make_shared<Fp64Scalar>(3.0),
                std::make_shared<Date32Scalar>(10),
                std::make_shared<Date64Scalar>(864000000),
                std::make_shared<Time32Scalar>(1000, time32(TimeUnit::SECOND)),
                std::make_shared<Time64Scalar>(1111, time64(TimeUnit::MICRO)),
                std::make_shared<TimestampScalar>(1111, timestamp(TimeUnit::MILLI)),
                std::make_shared<MonthIntervalScalar>(1),
                std::make_shared<DayTimeIntervalScalar>(daytime),
                std::make_shared<MonthDayNanoIntervalScalar>(month_day_nanos),
                std::make_shared<DurationScalar>(60, duration(TimeUnit::SECOND)),
                std::make_shared<BinaryScalar>(hello),
                std::make_shared<LargeBinaryScalar>(hello),
                std::make_shared<BinaryViewScalar>(hello),
                std::make_shared<FixedSizeBinaryScalar>(
                        hello, fixed_size_binary(static_cast<int32_t>(hello->size()))),
                std::make_shared<Decimal128Scalar>(Decimal128(10), decimal(16, 4)),
                std::make_shared<Decimal256Scalar>(Decimal256(10), decimal(76, 38)),
                std::make_shared<StringScalar>(hello),
                std::make_shared<LargeStringScalar>(hello),
                std::make_shared<StringViewScalar>(hello),
                std::make_shared<StringViewScalar>(Buffer::from_string("long string; not inlined")),
                std::make_shared<ListScalar>(assert_array_from_json(int8(), "[1, 2, 3]")),
                assert_scalar_from_json(map(int8(), utf8()), R"([[1, "foo"], [2, "bar"]])"),
                std::make_shared<LargeListScalar>(assert_array_from_json(int8(), "[1, 1, 2, 2, 3, 3]")),
                std::make_shared<FixedSizeListScalar>(assert_array_from_json(int8(), "[1, 2, 3, 4]")),
                std::make_shared<ListViewScalar>(assert_array_from_json(int8(), "[1, 2, 3]")),
                std::make_shared<LargeListViewScalar>(assert_array_from_json(int8(), "[1, 1, 2, 2, 3, 3]")),
                std::make_shared<StructScalar>(
                        ScalarVector{
                                std::make_shared<Int32Scalar>(2),
                                std::make_shared<Int32Scalar>(6),
                        },
                        STRUCT({field("min", int32()), field("max", int32())})),
                // Same values, different union type codes
                SparseUnionScalar::FromValue(std::make_shared<Int32Scalar>(100), 1,
                                             sparse_union_ty),
                SparseUnionScalar::FromValue(std::make_shared<Int32Scalar>(100), 2,
                                             sparse_union_ty),
                SparseUnionScalar::FromValue(MakeNullScalar(int32()), 2, sparse_union_ty),
                std::make_shared<DenseUnionScalar>(std::make_shared<Int32Scalar>(101), 6,
                                                   dense_union_ty),
                std::make_shared<DenseUnionScalar>(std::make_shared<Int32Scalar>(101), 42,
                                                   dense_union_ty),
                std::make_shared<DenseUnionScalar>(MakeNullScalar(int32()), 42, dense_union_ty),
                DictionaryScalar::create(assert_scalar_from_json(int8(), "1"),
                                         assert_array_from_json(utf8(), R"(["foo", "bar"])")),
                DictionaryScalar::create(assert_scalar_from_json(uint8(), "1"),
                                         assert_array_from_json(utf8(), R"(["foo", "bar"])")),
                std::make_shared<RunEndEncodedScalar>(assert_scalar_from_json(int8(), "1"),
                                                      run_end_encoded(int16(), int8())),
        };
    }

    TEST_F(TestArray, TestMakeArrayFromScalar) {
        ASSERT_OK_AND_ASSIGN(auto null_array, MakeArrayFromScalar(NullScalar(), 5));
        ASSERT_OK(null_array->validate_full());
        ASSERT_EQ(null_array->length(), 5);
        ASSERT_EQ(null_array->null_count(), 5);

        auto scalars = GetScalars();

        for (int64_t length: {16}) {
            for (auto scalar: scalars) {
                NEBULA_SCOPED_TRACE("scalar type: ", scalar->type->to_string());
                ASSERT_OK_AND_ASSIGN(auto array, MakeArrayFromScalar(*scalar, length));
                ASSERT_OK(array->validate_full());
                ASSERT_EQ(array->length(), length);
                ASSERT_EQ(array->null_count(), 0);
                ASSERT_EQ(array->device_type(), DeviceAllocationType::kCPU);

                // test case for ARROW-13321
                for (int64_t i: {int64_t{0}, length / 2, length - 1}) {
                    ASSERT_OK_AND_ASSIGN(auto s, array->get_scalar(i));
                    AssertScalarsEqual(*s, *scalar, /*verbose=*/true);
                }
            }
        }

        for (auto scalar: scalars) {
            NEBULA_SCOPED_TRACE("scalar type: ", scalar->type->to_string());
            AssertAppendScalar(pool_, scalar);
        }
    }

    TEST_F(TestArray, TestMakeArrayFromScalarSliced) {
        // Regression test for ARROW-13437
        auto scalars = GetScalars();

        for (auto scalar: scalars) {
            SCOPED_TRACE(scalar->type->to_string());
            ASSERT_OK_AND_ASSIGN(auto array, MakeArrayFromScalar(*scalar, 32));
            auto sliced = array->slice(1, 4);
            ASSERT_EQ(sliced->length(), 4);
            ASSERT_EQ(sliced->null_count(), 0);
            ASSERT_EQ(array->device_type(), DeviceAllocationType::kCPU);
            NEBULA_EXPECT_OK(sliced->validate_full());
        }
    }

    TEST_F(TestArray, TestMakeArrayFromDictionaryScalar) {
        auto dictionary = assert_array_from_json(utf8(), R"(["foo", "bar", "baz"])");
        auto type = std::make_shared<DictionaryType>(int8(), utf8());
        ASSERT_OK_AND_ASSIGN(auto value, MakeScalar(int8(), 1));
        auto scalar = DictionaryScalar({value, dictionary}, type);

        ASSERT_OK_AND_ASSIGN(auto array, MakeArrayFromScalar(scalar, 4));
        ASSERT_OK(array->validate_full());
        ASSERT_EQ(array->length(), 4);
        ASSERT_EQ(array->null_count(), 0);
        ASSERT_EQ(array->device_type(), DeviceAllocationType::kCPU);

        for (int i = 0; i < 4; i++) {
            ASSERT_OK_AND_ASSIGN(auto item, array->get_scalar(i));
            ASSERT_TRUE(item->equals(scalar));
        }
    }

    TEST_F(TestArray, TestMakeArrayFromMapScalar) {
        auto value =
                assert_array_from_json(STRUCT({field("key", utf8(), false), field("value", int8())}),
                              R"([{"key": "a", "value": 1}, {"key": "b", "value": 2}])");
        auto scalar = MapScalar(value);

        ASSERT_OK_AND_ASSIGN(auto array, MakeArrayFromScalar(scalar, 11));
        ASSERT_OK(array->validate_full());
        ASSERT_EQ(array->length(), 11);
        ASSERT_EQ(array->null_count(), 0);

        for (int i = 0; i < 11; i++) {
            ASSERT_OK_AND_ASSIGN(auto item, array->get_scalar(i));
            ASSERT_TRUE(item->equals(scalar));
        }

        AssertAppendScalar(pool_, std::make_shared<MapScalar>(scalar));
    }

    void CheckSpanRoundTrip(const Array &array) {
        ArraySpan span;
        span.set_members(*array.data());
        std::shared_ptr<Array> array2 = span.to_array();
        AssertArraysEqual(array, *array2);
    }

    TEST_F(TestArray, TestMakeEmptyArray) {
        for (auto type: TestArrayUtilitiesAgainstTheseTypes()) {
            NEBULA_SCOPED_TRACE("type = ", type->to_string());
            ASSERT_OK_AND_ASSIGN(auto array, MakeEmptyArray(type));
            ASSERT_OK(array->validate_full());
            ASSERT_EQ(array->length(), 0);

            CheckSpanRoundTrip(*array);
        }
    }

    TEST_F(TestArray, TestFillFromScalar) {
        for (auto type: TestArrayUtilitiesAgainstTheseTypes()) {
            NEBULA_SCOPED_TRACE("type = ", type->to_string());
            for (auto seed: {0u, 0xdeadbeef, 42u}) {
                NEBULA_SCOPED_TRACE("seed = ", seed);

                Field field("", type, /*nullable=*/true,
                            key_value_metadata({{"extension_allow_random_storage", "true"}}));
                auto array = random::GenerateArray(field, 1, seed);

                ASSERT_OK_AND_ASSIGN(auto scalar, array->get_scalar(0));

                ArraySpan span(*scalar);
                auto roundtripped_array = span.to_array();
                ASSERT_OK(roundtripped_array->validate_full());

                AssertArraysEqual(*array, *roundtripped_array);
                ASSERT_OK_AND_ASSIGN(auto roundtripped_scalar, roundtripped_array->get_scalar(0));
                AssertScalarsEqual(*scalar, *roundtripped_scalar);
            }
        }
    }

// GH-40069: Data-race when concurrent calling ArraySpan::FillFromScalar of the same
// scalar instance.
    TEST_F(TestArray, TestConcurrentFillFromScalar) {

        for (auto type: TestArrayUtilitiesAgainstTheseTypes()) {
            NEBULA_SCOPED_TRACE("type = ", type->to_string());
            for (auto seed: {0u, 0xdeadbeef, 42u}) {
                NEBULA_SCOPED_TRACE("seed = ", seed);

                Field field("", type, /*nullable=*/true,
                            key_value_metadata({{"extension_allow_random_storage", "true"}}));
                auto array = random::GenerateArray(field, 1, seed);

                ASSERT_OK_AND_ASSIGN(auto scalar, array->get_scalar(0));

                // Lambda to create fill an ArraySpan with the scalar and use the ArraySpan a bit.
                auto array_span_from_scalar = [&]() {
                    ArraySpan span(*scalar);
                    auto roundtripped_array = span.to_array();
                    ASSERT_OK(roundtripped_array->validate_full());

                    AssertArraysEqual(*array, *roundtripped_array);
                    ASSERT_OK_AND_ASSIGN(auto roundtripped_scalar, roundtripped_array->get_scalar(0));
                    AssertScalarsEqual(*scalar, *roundtripped_scalar);
                };

                // Two concurrent calls to the lambda are just enough for TSAN to detect a race
                // condition.
                auto fut1 = std::async(std::launch::async, array_span_from_scalar);
                auto fut2 = std::async(std::launch::async, array_span_from_scalar);
                fut1.get();
                fut2.get();
            }
        }
    }

    TEST_F(TestArray, ExtensionSpanRoundTrip) {
        // Other types are checked in MakeEmptyArray but MakeEmptyArray doesn't
        // work for extension types so we check that here
        ASSERT_OK_AND_ASSIGN(auto array, MakeEmptyArray(dictionary(int8(), utf8())));
        auto ext_array = ExtensionArray(dict_extension_type(), std::move(array));
        CheckSpanRoundTrip(ext_array);
    }

    TEST_F(TestArray, TestAppendArraySlice) {
        auto scalars = GetScalars();
        ArraySpan span;
        for (const auto &scalar: scalars) {
            NEBULA_SCOPED_TRACE(*scalar->type);
            ASSERT_OK_AND_ASSIGN(auto array, MakeArrayFromScalar(*scalar, 16));
            ASSERT_OK_AND_ASSIGN(auto nulls, MakeArrayOfNull(scalar->type, 16));

            std::unique_ptr<nebula::ArrayBuilder> builder;
            ASSERT_OK(MakeBuilder(pool_, scalar->type, &builder));

            span.set_members(*array->data());
            ASSERT_OK(builder->append_array_slice(span, 0, 4));
            ASSERT_EQ(4, builder->length());
            ASSERT_OK(builder->append_array_slice(span, 0, 0));
            ASSERT_EQ(4, builder->length());
            ASSERT_OK(builder->append_array_slice(span, 1, 0));
            ASSERT_EQ(4, builder->length());
            ASSERT_OK(builder->append_array_slice(span, 1, 4));
            ASSERT_EQ(8, builder->length());

            span.set_members(*nulls->data());
            ASSERT_OK(builder->append_array_slice(span, 0, 4));
            ASSERT_EQ(12, builder->length());
            const bool has_validity_bitmap =
                    internal::may_have_validity_bitmap(scalar->type->id());
            if (has_validity_bitmap) {
                ASSERT_EQ(4, builder->null_count());
            }
            ASSERT_OK(builder->append_array_slice(span, 0, 0));
            ASSERT_EQ(12, builder->length());
            if (has_validity_bitmap) {
                ASSERT_EQ(4, builder->null_count());
            }
            ASSERT_OK(builder->append_array_slice(span, 1, 0));
            ASSERT_EQ(12, builder->length());
            if (has_validity_bitmap) {
                ASSERT_EQ(4, builder->null_count());
            }
            ASSERT_OK(builder->append_array_slice(span, 1, 4));
            ASSERT_EQ(16, builder->length());
            if (has_validity_bitmap) {
                ASSERT_EQ(8, builder->null_count());
            }

            std::shared_ptr<Array> result;
            ASSERT_OK(builder->finish(&result));
            ASSERT_OK(result->validate_full());
            ASSERT_EQ(16, result->length());
            if (has_validity_bitmap) {
                ASSERT_EQ(8, result->null_count());
            }
        }

        {
            ASSERT_OK_AND_ASSIGN(auto array, MakeArrayOfNull(null(), 16));
            NullBuilder builder(pool_);

            span.set_members(*array->data());
            ASSERT_OK(builder.append_array_slice(span, 0, 4));
            ASSERT_EQ(4, builder.length());
            ASSERT_OK(builder.append_array_slice(span, 0, 0));
            ASSERT_EQ(4, builder.length());
            ASSERT_OK(builder.append_array_slice(span, 1, 0));
            ASSERT_EQ(4, builder.length());
            ASSERT_OK(builder.append_array_slice(span, 1, 4));
            ASSERT_EQ(8, builder.length());
            std::shared_ptr<Array> result;
            ASSERT_OK(builder.finish(&result));
            ASSERT_OK(result->validate_full());
            ASSERT_EQ(8, result->length());
            ASSERT_EQ(8, result->null_count());
        }
    }

// GH-39976: Test out-of-line data size calculation in
// BinaryViewBuilder::append_array_slice.
    TEST_F(TestArray, TestBinaryViewAppendArraySlice) {
        BinaryViewBuilder src_builder(pool_);
        ASSERT_OK(src_builder.append_null());
        ASSERT_OK(src_builder.append("long string; not inlined"));
        ASSERT_EQ(2, src_builder.length());
        ASSERT_OK_AND_ASSIGN(auto src, src_builder.finish());
        ASSERT_OK(src->validate_full());

        ArraySpan span;
        span.set_members(*src->data());
        BinaryViewBuilder dst_builder(pool_);
        ASSERT_OK(dst_builder.append_array_slice(span, 0, 1));
        ASSERT_EQ(1, dst_builder.length());
        ASSERT_OK(dst_builder.append_array_slice(span, 1, 1));
        ASSERT_EQ(2, dst_builder.length());
        ASSERT_OK_AND_ASSIGN(auto dst, dst_builder.finish());
        ASSERT_OK(dst->validate_full());

        AssertArraysEqual(*src, *dst);
    }

    TEST_F(TestArray, ValidateBuffersPrimitive) {
        auto empty_buffer = std::make_shared<Buffer>("");
        auto null_buffer = Buffer::from_string("\xff");
        auto data_buffer = Buffer::from_string("123456789abcdef0");

        auto data = ArrayData::create(int64(), 2, {null_buffer, data_buffer});
        auto array = make_array(data);
        ASSERT_OK(array->validate_full());
        data = ArrayData::create(boolean(), 8, {null_buffer, data_buffer});
        array = make_array(data);
        ASSERT_OK(array->validate_full());

        // Null buffer too small
        data = ArrayData::create(int64(), 2, {empty_buffer, data_buffer});
        array = make_array(data);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, array->validate());
        data = ArrayData::create(boolean(), 9, {null_buffer, data_buffer});
        array = make_array(data);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, array->validate());

        // Data buffer too small
        data = ArrayData::create(int64(), 3, {null_buffer, data_buffer});
        array = make_array(data);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, array->validate());

        // Null buffer absent but null_count > 0.
        data = ArrayData::create(int64(), 2, {nullptr, data_buffer}, 1);
        array = make_array(data);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, array->validate());

        //
        // With offset > 0
        //
        data = ArrayData::create(int64(), 1, {null_buffer, data_buffer}, kUnknownNullCount, 1);
        array = make_array(data);
        ASSERT_OK(array->validate_full());
        data = ArrayData::create(boolean(), 6, {null_buffer, data_buffer}, kUnknownNullCount, 2);
        array = make_array(data);
        ASSERT_OK(array->validate_full());

        // Null buffer too small
        data = ArrayData::create(boolean(), 7, {null_buffer, data_buffer}, kUnknownNullCount, 2);
        array = make_array(data);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, array->validate());

        // Data buffer too small
        data = ArrayData::create(int64(), 2, {null_buffer, data_buffer}, kUnknownNullCount, 1);
        array = make_array(data);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, array->validate());
    }

// ----------------------------------------------------------------------
// Null type tests

    TEST(TestNullBuilder, Basics) {
        NullBuilder builder;
        std::shared_ptr<Array> array;

        ASSERT_OK(builder.append_null());
        ASSERT_OK(builder.append(nullptr));
        ASSERT_OK(builder.append_null());
        ASSERT_OK(builder.append_nulls(2));
        ASSERT_EQ(5, builder.null_count());
        ASSERT_OK(builder.finish(&array));

        const auto &null_array = turbo::checked_cast<NullArray &>(*array);
        ASSERT_EQ(null_array.length(), 5);
        ASSERT_EQ(null_array.null_count(), 5);
    }

// ----------------------------------------------------------------------
// Primitive type tests

    TEST(TestPrimitiveArray, CtorNoValidityBitmap) {
        // ARROW-8863
        std::shared_ptr<Buffer> data = *allocate_buffer(40);
        Int32Array arr(10, data);
        ASSERT_EQ(arr.data()->null_count, 0);
    }

    class TestBuilder : public ::testing::Test {
    protected:
        MemoryPool *pool_ = default_memory_pool();
        std::shared_ptr<DataType> type_;
    };

    TEST_F(TestBuilder, TestReserve) {
        UInt8Builder builder(pool_);

        ASSERT_OK(builder.resize(1000));
        ASSERT_EQ(1000, builder.capacity());

        // Reserve overallocates for small upsizes.
        ASSERT_OK(builder.Reserve(1030));
        ASSERT_GE(builder.capacity(), 2000);
    }

    TEST_F(TestBuilder, TestResizeDownsize) {
        UInt8Builder builder(pool_);

        ASSERT_OK(builder.resize(1000));
        ASSERT_EQ(1000, builder.capacity());
        ASSERT_EQ(0, builder.length());
        ASSERT_OK(builder.append_nulls(500));
        ASSERT_EQ(1000, builder.capacity());
        ASSERT_EQ(500, builder.length());

        // Can downsize below current capacity
        ASSERT_OK(builder.resize(500));
        // ... but not below current populated length
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, builder.resize(499));
        ASSERT_GE(500, builder.capacity());
        ASSERT_EQ(500, builder.length());
    }

    template<typename Attrs>
    class TestPrimitiveBuilder : public TestBuilder {
    public:
        typedef Attrs TestAttrs;
        typedef typename Attrs::ArrayType ArrayType;
        typedef typename Attrs::BuilderType BuilderType;
        typedef typename Attrs::T CType;
        typedef typename Attrs::Type Type;

        virtual void SetUp() {
            TestBuilder::SetUp();

            type_ = Attrs::type();

            std::unique_ptr<ArrayBuilder> tmp;
            ASSERT_OK(MakeBuilder(pool_, type_, &tmp));
            builder_.reset(turbo::checked_cast<BuilderType *>(tmp.release()));

            ASSERT_OK(MakeBuilder(pool_, type_, &tmp));
            builder_nn_.reset(turbo::checked_cast<BuilderType *>(tmp.release()));
        }

        void RandomData(int64_t N, double pct_null = 0.1) {
            Attrs::draw(N, &draws_);

            valid_bytes_.resize(static_cast<size_t>(N));
            random_null_bytes(N, pct_null, valid_bytes_.data());
        }

        void Check(const std::unique_ptr<BuilderType> &builder, bool nullable) {
            int64_t size = builder->length();
            auto ex_data = Buffer::Wrap(draws_.data(), size);

            std::shared_ptr<Buffer> ex_null_bitmap;
            int64_t ex_null_count = 0;

            if (nullable) {
                ASSERT_OK_AND_ASSIGN(ex_null_bitmap, internal::BytesToBits(valid_bytes_));
                ex_null_count = CountNulls(valid_bytes_);
            } else {
                ex_null_bitmap = nullptr;
            }

            auto expected =
                    std::make_shared<ArrayType>(size, ex_data, ex_null_bitmap, ex_null_count);

            std::shared_ptr<Array> out;
            FinishAndCheckPadding(builder.get(), &out);

            std::shared_ptr<ArrayType> result = turbo::checked_pointer_cast<ArrayType>(out);

            // Builder is now reset
            ASSERT_EQ(0, builder->length());
            ASSERT_EQ(0, builder->capacity());
            ASSERT_EQ(0, builder->null_count());

            ASSERT_EQ(ex_null_count, result->null_count());
            ASSERT_TRUE(result->equals(*expected));
        }

        void FlipValue(CType *ptr) {
            auto byteptr = reinterpret_cast<uint8_t *>(ptr);
            *byteptr = static_cast<uint8_t>(~*byteptr);
        }

    protected:
        std::unique_ptr<BuilderType> builder_;
        std::unique_ptr<BuilderType> builder_nn_;

        std::vector<CType> draws_;
        std::vector<uint8_t> valid_bytes_;
    };

/// \brief uint8_t isn't a valid template parameter to uniform_int_distribution, so
/// we use SampleType to determine which kind of integer to use to sample.
    template<typename T, typename = enable_if_t<std::is_integral<T>::value, T>>
    struct UniformIntSampleType {
        using type = T;
    };

    template<>
    struct UniformIntSampleType<uint8_t> {
        using type = uint16_t;
    };

    template<>
    struct UniformIntSampleType<int8_t> {
        using type = int16_t;
    };

#define PTYPE_DECL(CapType, c_type)     \
  typedef CapType##Array ArrayType;     \
  typedef CapType##Builder BuilderType; \
  typedef CapType##Type Type;           \
  typedef c_type T;                     \
                                        \
  static std::shared_ptr<DataType> type() { return std::make_shared<Type>(); }

#define PINT_DECL(CapType, c_type)                                                     \
  struct P##CapType {                                                                  \
    PTYPE_DECL(CapType, c_type)                                                        \
    static void draw(int64_t N, std::vector<T>* draws) {                               \
      using sample_type = typename UniformIntSampleType<c_type>::type;                 \
      const T lower = std::numeric_limits<T>::min();                                   \
      const T upper = std::numeric_limits<T>::max();                                   \
      randint(N, static_cast<sample_type>(lower), static_cast<sample_type>(upper),     \
              draws);                                                                  \
    }                                                                                  \
    static T Modify(T inp) { return inp / 2; }                                         \
    typedef                                                                            \
        typename std::conditional<std::is_unsigned<T>::value, uint64_t, int64_t>::type \
            ConversionType;                                                            \
  }

#define PFLOAT_DECL(CapType, c_type, LOWER, UPPER)       \
  struct P##CapType {                                    \
    PTYPE_DECL(CapType, c_type)                          \
    static void draw(int64_t N, std::vector<T>* draws) { \
      random_real(N, 0, LOWER, UPPER, draws);            \
    }                                                    \
    static T Modify(T inp) { return inp / 2; }           \
    typedef double ConversionType;                       \
  }

    PINT_DECL(UInt8, uint8_t);

    PINT_DECL(UInt16, uint16_t);

    PINT_DECL(UInt32, uint32_t);

    PINT_DECL(UInt64, uint64_t);

    PINT_DECL(Int8, int8_t);

    PINT_DECL(Int16, int16_t);

    PINT_DECL(Int32, int32_t);

    PINT_DECL(Int64, int64_t);

    PFLOAT_DECL(Fp32, float, -1000.0f, 1000.0f);

    PFLOAT_DECL(Fp64, double, -1000.0, 1000.0);

    struct PBoolean {
        PTYPE_DECL(Boolean, uint8_t)

        static T Modify(T inp) { return !inp; }

        typedef int64_t ConversionType;
    };

    struct PDayTimeInterval {
        using DayMilliseconds = DayTimeIntervalType::DayMilliseconds;
        PTYPE_DECL(DayTimeInterval, DayMilliseconds);

        static void draw(int64_t N, std::vector<T> *draws) { return rand_day_millis(N, draws); }

        static DayMilliseconds Modify(DayMilliseconds inp) {
            inp.days /= 2;
            return inp;
        }

        typedef DayMilliseconds ConversionType;
    };

    struct PMonthDayNanoInterval {
        using MonthDayNanos = MonthDayNanoIntervalType::MonthDayNanos;
        PTYPE_DECL(MonthDayNanoInterval, MonthDayNanos);

        static void draw(int64_t N, std::vector<T> *draws) {
            return rand_month_day_nanos(N, draws);
        }

        static MonthDayNanos Modify(MonthDayNanos inp) {
            inp.days /= 2;
            return inp;
        }

        typedef MonthDayNanos ConversionType;
    };

    template<>
    void TestPrimitiveBuilder<PBoolean>::RandomData(int64_t N, double pct_null) {
        draws_.resize(static_cast<size_t>(N));
        valid_bytes_.resize(static_cast<size_t>(N));

        random_null_bytes(N, 0.5, draws_.data());
        random_null_bytes(N, pct_null, valid_bytes_.data());
    }

    template<>
    void TestPrimitiveBuilder<PBoolean>::FlipValue(CType *ptr) {
        *ptr = !*ptr;
    }

    template<>
    void TestPrimitiveBuilder<PBoolean>::Check(const std::unique_ptr<BooleanBuilder> &builder,
                                               bool nullable) {
        const int64_t size = builder->length();

        // Build expected result array
        std::shared_ptr<Buffer> ex_data;
        std::shared_ptr<Buffer> ex_null_bitmap;
        int64_t ex_null_count = 0;

        ASSERT_OK_AND_ASSIGN(ex_data, internal::BytesToBits(draws_));
        if (nullable) {
            ASSERT_OK_AND_ASSIGN(ex_null_bitmap, internal::BytesToBits(valid_bytes_));
            ex_null_count = CountNulls(valid_bytes_);
        } else {
            ex_null_bitmap = nullptr;
        }
        auto expected =
                std::make_shared<BooleanArray>(size, ex_data, ex_null_bitmap, ex_null_count);
        ASSERT_EQ(size, expected->length());

        // finish builder and check result array
        std::shared_ptr<Array> out;
        FinishAndCheckPadding(builder.get(), &out);

        std::shared_ptr<BooleanArray> result = turbo::checked_pointer_cast<BooleanArray>(out);

        ASSERT_EQ(ex_null_count, result->null_count());
        ASSERT_EQ(size, result->length());

        for (int64_t i = 0; i < size; ++i) {
            if (nullable) {
                ASSERT_EQ(valid_bytes_[i] == 0, result->is_null(i)) << i;
            } else {
                ASSERT_FALSE(result->is_null(i));
            }
            if (!result->is_null(i)) {
                bool actual = bit_util::get_bit(result->values()->data(), i);
                ASSERT_EQ(draws_[i] != 0, actual) << i;
            }
        }
        AssertArraysEqual(*result, *expected);

        // buffers are correctly sized
        if (result->data()->buffers[0]) {
            ASSERT_EQ(result->data()->buffers[0]->size(), bit_util::BytesForBits(size));
        } else {
            ASSERT_EQ(result->data()->null_count, 0);
        }
        ASSERT_EQ(result->data()->buffers[1]->size(), bit_util::BytesForBits(size));

        // Builder is now reset
        ASSERT_EQ(0, builder->length());
        ASSERT_EQ(0, builder->capacity());
        ASSERT_EQ(0, builder->null_count());
    }

    TEST(TestBooleanArray, TrueCountFalseCount) {
        random::RandomArrayGenerator rng(/*seed=*/0);

        const int64_t length = 10000;
        auto arr = rng.Boolean(length, /*true_probability=*/0.5, /*null_probability=*/0.1);

        auto CheckArray = [&](const BooleanArray &values) {
            int64_t expected_false = 0;
            int64_t expected_true = 0;
            for (int64_t i = 0; i < values.length(); ++i) {
                if (values.is_valid(i)) {
                    if (values.value(i)) {
                        ++expected_true;
                    } else {
                        ++expected_false;
                    }
                }
            }
            ASSERT_EQ(values.true_count(), expected_true);
            ASSERT_EQ(values.false_count(), expected_false);
        };

        CheckArray(turbo::checked_cast<const BooleanArray &>(*arr));
        CheckArray(turbo::checked_cast<const BooleanArray &>(*arr->slice(5)));
        CheckArray(turbo::checked_cast<const BooleanArray &>(*arr->slice(0, 0)));

        // GH-41016 true_count() with array without validity buffer with null_count of -1
        auto arr_unknown_null_count = assert_array_from_json(boolean(), "[true, false, true]");
        arr_unknown_null_count->data()->null_count = kUnknownNullCount;
        ASSERT_EQ(arr_unknown_null_count->data()->null_count.load(), -1);
        ASSERT_EQ(arr_unknown_null_count->null_bitmap(), nullptr);
        ASSERT_EQ(turbo::checked_pointer_cast<BooleanArray>(arr_unknown_null_count)->true_count(), 2);
    }

    TEST(TestPrimitiveAdHoc, TestType) {
        Int8Builder i8(default_memory_pool());
        ASSERT_TRUE(i8.type()->equals(int8()));

        DictionaryBuilder<Int8Type> d_i8(utf8());
        ASSERT_TRUE(d_i8.type()->equals(dictionary(int8(), utf8())));

        Dictionary32Builder<Int8Type> d32_i8(utf8());
        ASSERT_TRUE(d32_i8.type()->equals(dictionary(int32(), utf8())));
    }

    TEST(NumericBuilderAccessors, TestSettersGetters) {
        int64_t datum = 42;
        int64_t new_datum = 43;
        NumericBuilder<Int64Type> builder(int64(), default_memory_pool());

        builder.reset();
        ASSERT_OK(builder.append(datum));
        ASSERT_EQ(builder.get_value(0), datum);

        // Now update the value.
        builder[0] = new_datum;

        ASSERT_EQ(builder.get_value(0), new_datum);
        ASSERT_EQ(((const NumericBuilder<Int64Type> &) builder)[0], new_datum);
    }

    typedef ::testing::Types<PBoolean, PUInt8, PUInt16, PUInt32, PUInt64, PInt8, PInt16,
            PInt32, PInt64, PFp32, PFp64, PDayTimeInterval,
            PMonthDayNanoInterval>
            Primitives;

    TYPED_TEST_SUITE(TestPrimitiveBuilder, Primitives);

    TYPED_TEST(TestPrimitiveBuilder, TestInit) {
        ASSERT_OK(this->builder_->Reserve(1000));
        ASSERT_EQ(1000, this->builder_->capacity());

        // Small upsize => should overallocate
        ASSERT_OK(this->builder_->Reserve(1200));
        ASSERT_GE(2000, this->builder_->capacity());

        // Large upsize => should allocate exactly
        ASSERT_OK(this->builder_->Reserve(32768));
        ASSERT_EQ(32768, this->builder_->capacity());

        // unsure if this should go in all builder classes
        ASSERT_EQ(0, this->builder_->num_children());
    }

    TYPED_TEST(TestPrimitiveBuilder, TestAppendNull) {
        int64_t size = 1000;
        for (int64_t i = 0; i < size; ++i) {
            ASSERT_OK(this->builder_->append_null());
            ASSERT_EQ(i + 1, this->builder_->null_count());
        }

        std::shared_ptr<Array> out;
        FinishAndCheckPadding(this->builder_.get(), &out);
        auto result = turbo::checked_pointer_cast<typename TypeParam::ArrayType>(out);

        for (int64_t i = 0; i < size; ++i) {
            ASSERT_TRUE(result->is_null(i)) << i;
        }

        for (auto buffer: result->data()->buffers) {
            for (int64_t i = 0; i < buffer->capacity(); i++) {
                // Validates current implementation, algorithms shouldn't rely on this
                ASSERT_EQ(0, *(buffer->data() + i)) << i;
            }
        }
    }

    TYPED_TEST(TestPrimitiveBuilder, TestAppendOptional) {
        int64_t size = 1000;
        for (int64_t i = 0; i < size; ++i) {
            ASSERT_OK(this->builder_->AppendOrNull(std::nullopt));
            ASSERT_EQ(i + 1, this->builder_->null_count());
        }

        std::shared_ptr<Array> out;
        FinishAndCheckPadding(this->builder_.get(), &out);
        auto result = turbo::checked_pointer_cast<typename TypeParam::ArrayType>(out);

        for (int64_t i = 0; i < size; ++i) {
            ASSERT_TRUE(result->is_null(i)) << i;
        }
    }

    TYPED_TEST(TestPrimitiveBuilder, TestAppendNulls) {
        const int64_t size = 10;
        ASSERT_OK(this->builder_->append_nulls(size));
        ASSERT_EQ(size, this->builder_->null_count());

        std::shared_ptr<Array> result;
        FinishAndCheckPadding(this->builder_.get(), &result);

        for (int64_t i = 0; i < size; ++i) {
            ASSERT_FALSE(result->is_valid(i));
        }

        for (auto buffer: result->data()->buffers) {
            for (int64_t i = 0; i < buffer->capacity(); i++) {
                // Validates current implementation, algorithms shouldn't rely on this
                ASSERT_EQ(0, *(buffer->data() + i)) << i;
            }
        }
    }

    TYPED_TEST(TestPrimitiveBuilder, TestAppendEmptyValue) {
        ASSERT_OK(this->builder_->append_null());
        ASSERT_OK(this->builder_->append_empty_value());
        ASSERT_OK(this->builder_->append_nulls(2));
        ASSERT_OK(this->builder_->append_empty_values(2));

        std::shared_ptr<Array> out;
        FinishAndCheckPadding(this->builder_.get(), &out);
        ASSERT_OK(out->validate_full());

        auto result = turbo::checked_pointer_cast<typename TypeParam::ArrayType>(out);
        ASSERT_EQ(result->length(), 6);
        ASSERT_EQ(result->null_count(), 3);

        ASSERT_TRUE(result->is_null(0));
        ASSERT_FALSE(result->is_null(1));
        ASSERT_TRUE(result->is_null(2));
        ASSERT_TRUE(result->is_null(3));
        ASSERT_FALSE(result->is_null(4));
        ASSERT_FALSE(result->is_null(5));

        // implementation detail: the value slots are 0-initialized
        for (int64_t i = 0; i < result->length(); ++i) {
            typename TestFixture::CType t{};
            ASSERT_EQ(result->value(i), t);
        }
    }

    TYPED_TEST(TestPrimitiveBuilder, TestArrayDtorDealloc) {
        typedef typename TestFixture::CType T;

        int64_t size = 1000;

        std::vector<T> &draws = this->draws_;
        std::vector<uint8_t> &valid_bytes = this->valid_bytes_;

        int64_t memory_before = this->pool_->bytes_allocated();

        this->RandomData(size);
        ASSERT_OK(this->builder_->Reserve(size));

        int64_t i;
        for (i = 0; i < size; ++i) {
            if (valid_bytes[i] > 0) {
                ASSERT_OK(this->builder_->append(draws[i]));
            } else {
                ASSERT_OK(this->builder_->append_null());
            }
        }

        do {
            std::shared_ptr<Array> result;
            FinishAndCheckPadding(this->builder_.get(), &result);
        } while (false);

        ASSERT_EQ(memory_before, this->pool_->bytes_allocated());
    }

    TYPED_TEST(TestPrimitiveBuilder, Equality) {
        typedef typename TestFixture::CType T;

        const int64_t size = 1000;
        this->RandomData(size);
        std::vector<T> &draws = this->draws_;
        std::vector<uint8_t> &valid_bytes = this->valid_bytes_;
        std::shared_ptr<Array> array, equal_array, unequal_array;
        auto builder = this->builder_.get();
        ASSERT_OK(make_array(valid_bytes, draws, size, builder, &array));
        ASSERT_OK(make_array(valid_bytes, draws, size, builder, &equal_array));

        // Make the not equal array by negating the first valid element with itself.
        const auto first_valid = std::find_if(valid_bytes.begin(), valid_bytes.end(),
                                              [](uint8_t valid) { return valid > 0; });
        const int64_t first_valid_idx = std::distance(valid_bytes.begin(), first_valid);
        // This should be true with a very high probability, but might introduce flakiness
        ASSERT_LT(first_valid_idx, size - 1);
        this->FlipValue(&draws[first_valid_idx]);
        ASSERT_OK(make_array(valid_bytes, draws, size, builder, &unequal_array));

        // test normal equality
        EXPECT_TRUE(array->equals(array));
        EXPECT_TRUE(array->equals(equal_array));
        EXPECT_TRUE(equal_array->equals(array));
        EXPECT_FALSE(equal_array->equals(unequal_array));
        EXPECT_FALSE(unequal_array->equals(equal_array));

        // Test range equality
        EXPECT_FALSE(array->range_equals(0, first_valid_idx + 1, 0, unequal_array));
        EXPECT_FALSE(array->range_equals(first_valid_idx, size, first_valid_idx, unequal_array));
        EXPECT_TRUE(array->range_equals(0, first_valid_idx, 0, unequal_array));
        EXPECT_TRUE(
                array->range_equals(first_valid_idx + 1, size, first_valid_idx + 1, unequal_array));
    }

    TYPED_TEST(TestPrimitiveBuilder, SliceEquality) {
        typedef typename TestFixture::CType T;

        const int64_t size = 1000;
        this->RandomData(size);
        std::vector<T> &draws = this->draws_;
        std::vector<uint8_t> &valid_bytes = this->valid_bytes_;
        auto builder = this->builder_.get();

        std::shared_ptr<Array> array;
        ASSERT_OK(make_array(valid_bytes, draws, size, builder, &array));

        std::shared_ptr<Array> slice, slice2;

        slice = array->slice(5);
        slice2 = array->slice(5);
        ASSERT_EQ(size - 5, slice->length());

        ASSERT_TRUE(slice->equals(slice2));
        ASSERT_TRUE(array->range_equals(5, array->length(), 0, slice));

        // Chained slices
        slice2 = array->slice(2)->slice(3);
        ASSERT_TRUE(slice->equals(slice2));

        slice = array->slice(5, 10);
        slice2 = array->slice(5, 10);
        ASSERT_EQ(10, slice->length());

        ASSERT_TRUE(slice->equals(slice2));
        ASSERT_TRUE(array->range_equals(5, 15, 0, slice));
    }

    TYPED_TEST(TestPrimitiveBuilder, TestAppendScalar) {
        typedef typename TestFixture::CType T;

        const int64_t size = 10000;

        std::vector<T> &draws = this->draws_;
        std::vector<uint8_t> &valid_bytes = this->valid_bytes_;

        this->RandomData(size);

        ASSERT_OK(this->builder_->Reserve(1000));
        ASSERT_OK(this->builder_nn_->Reserve(1000));

        int64_t null_count = 0;
        // append the first 1000
        for (size_t i = 0; i < 1000; ++i) {
            if (valid_bytes[i] > 0) {
                ASSERT_OK(this->builder_->append(draws[i]));
            } else {
                ASSERT_OK(this->builder_->append_null());
                ++null_count;
            }
            ASSERT_OK(this->builder_nn_->append(draws[i]));
        }

        ASSERT_EQ(null_count, this->builder_->null_count());

        ASSERT_EQ(1000, this->builder_->length());
        ASSERT_EQ(1000, this->builder_->capacity());

        ASSERT_EQ(1000, this->builder_nn_->length());
        ASSERT_EQ(1000, this->builder_nn_->capacity());

        ASSERT_OK(this->builder_->Reserve(size - 1000));
        ASSERT_OK(this->builder_nn_->Reserve(size - 1000));

        // append the next 9000
        for (size_t i = 1000; i < size; ++i) {
            if (valid_bytes[i] > 0) {
                ASSERT_OK(this->builder_->append(draws[i]));
            } else {
                ASSERT_OK(this->builder_->append_null());
            }
            ASSERT_OK(this->builder_nn_->append(draws[i]));
        }

        ASSERT_EQ(size, this->builder_->length());
        ASSERT_GE(size, this->builder_->capacity());

        ASSERT_EQ(size, this->builder_nn_->length());
        ASSERT_GE(size, this->builder_nn_->capacity());

        this->Check(this->builder_, true);
        this->Check(this->builder_nn_, false);
    }

    TYPED_TEST(TestPrimitiveBuilder, TestAppendValues) {
        typedef typename TestFixture::CType T;

        int64_t size = 10000;
        this->RandomData(size);

        std::vector<T> &draws = this->draws_;
        std::vector<uint8_t> &valid_bytes = this->valid_bytes_;

        // first slug
        int64_t K = 1000;

        ASSERT_OK(this->builder_->append_values(draws.data(), K, valid_bytes.data()));
        ASSERT_OK(this->builder_nn_->append_values(draws.data(), K));

        ASSERT_EQ(1000, this->builder_->length());
        ASSERT_EQ(1000, this->builder_->capacity());

        ASSERT_EQ(1000, this->builder_nn_->length());
        ASSERT_EQ(1000, this->builder_nn_->capacity());

        // append the next 9000
        ASSERT_OK(
                this->builder_->append_values(draws.data() + K, size - K, valid_bytes.data() + K));
        ASSERT_OK(this->builder_nn_->append_values(draws.data() + K, size - K));

        ASSERT_EQ(size, this->builder_->length());
        ASSERT_GE(size, this->builder_->capacity());

        ASSERT_EQ(size, this->builder_nn_->length());
        ASSERT_GE(size, this->builder_nn_->capacity());

        this->Check(this->builder_, true);
        this->Check(this->builder_nn_, false);
    }

    TYPED_TEST(TestPrimitiveBuilder, TestTypedFinish) {
        typedef typename TestFixture::CType T;

        int64_t size = 1000;
        this->RandomData(size);

        std::vector<T> &draws = this->draws_;
        std::vector<uint8_t> &valid_bytes = this->valid_bytes_;

        ASSERT_OK(this->builder_->append_values(draws.data(), size, valid_bytes.data()));
        std::shared_ptr<Array> result_untyped;
        ASSERT_OK(this->builder_->finish(&result_untyped));

        ASSERT_OK(this->builder_->append_values(draws.data(), size, valid_bytes.data()));
        std::shared_ptr<typename TestFixture::ArrayType> result;
        ASSERT_OK(this->builder_->finish(&result));

        AssertArraysEqual(*result_untyped, *result);
    }

    TYPED_TEST(TestPrimitiveBuilder, TestAppendValuesIter) {
        int64_t size = 10000;
        this->RandomData(size);

        ASSERT_OK(this->builder_->append_values(this->draws_.begin(), this->draws_.end(),
                                               this->valid_bytes_.begin()));
        ASSERT_OK(this->builder_nn_->append_values(this->draws_.begin(), this->draws_.end()));

        ASSERT_EQ(size, this->builder_->length());
        ASSERT_GE(size, this->builder_->capacity());

        this->Check(this->builder_, true);
        this->Check(this->builder_nn_, false);
    }

    TYPED_TEST(TestPrimitiveBuilder, TestAppendValuesIterNullValid) {
        int64_t size = 10000;
        this->RandomData(size);

        ASSERT_OK(this->builder_nn_->append_values(this->draws_.begin(),
                                                  this->draws_.begin() + size / 2,
                                                  static_cast<uint8_t *>(nullptr)));

        ASSERT_GE(size / 2, this->builder_nn_->capacity());

        ASSERT_OK(this->builder_nn_->append_values(this->draws_.begin() + size / 2,
                                                  this->draws_.end(),
                                                  static_cast<uint64_t *>(nullptr)));

        this->Check(this->builder_nn_, false);
    }

    TYPED_TEST(TestPrimitiveBuilder, TestAppendValuesLazyIter) {
        typedef typename TestFixture::CType T;

        int64_t size = 10000;
        this->RandomData(size);

        auto &draws = this->draws_;
        auto &valid_bytes = this->valid_bytes_;

        auto halve = [&draws](int64_t index) {
            return TestFixture::TestAttrs::Modify(draws[index]);
        };
        auto lazy_iter = internal::MakeLazyRange(halve, size);

        ASSERT_OK(this->builder_->append_values(lazy_iter.begin(), lazy_iter.end(),
                                               valid_bytes.begin()));

        std::vector<T> halved;
        transform(draws.begin(), draws.end(), back_inserter(halved),
                  [](T in) { return TestFixture::TestAttrs::Modify(in); });

        std::shared_ptr<Array> result;
        FinishAndCheckPadding(this->builder_.get(), &result);

        std::shared_ptr<Array> expected;
        ASSERT_OK(
                this->builder_->append_values(halved.data(), halved.size(), valid_bytes.data()));
        FinishAndCheckPadding(this->builder_.get(), &expected);

        ASSERT_TRUE(expected->equals(result));
    }

    TYPED_TEST(TestPrimitiveBuilder, TestAppendValuesIterConverted) {
        typedef typename TestFixture::CType T;
        // find type we can safely convert the tested values to and from
        using conversion_type = typename TestFixture::TestAttrs::ConversionType;

        int64_t size = 10000;
        this->RandomData(size);

        // append convertible values
        std::vector<conversion_type> draws_converted(this->draws_.begin(), this->draws_.end());
        std::vector<int32_t> valid_bytes_converted(this->valid_bytes_.begin(),
                                                   this->valid_bytes_.end());

        auto cast_values = internal::MakeLazyRange(
                [&draws_converted](int64_t index) {
                    return static_cast<T>(draws_converted[index]);
                },
                size);
        auto cast_valid = internal::MakeLazyRange(
                [&valid_bytes_converted](int64_t index) {
                    return static_cast<bool>(valid_bytes_converted[index]);
                },
                size);

        ASSERT_OK(this->builder_->append_values(cast_values.begin(), cast_values.end(),
                                               cast_valid.begin()));
        ASSERT_OK(this->builder_nn_->append_values(cast_values.begin(), cast_values.end()));

        ASSERT_EQ(size, this->builder_->length());
        ASSERT_GE(size, this->builder_->capacity());

        ASSERT_EQ(size, this->builder_->length());
        ASSERT_GE(size, this->builder_->capacity());

        this->Check(this->builder_, true);
        this->Check(this->builder_nn_, false);
    }

    TYPED_TEST(TestPrimitiveBuilder, TestZeroPadded) {
        typedef typename TestFixture::CType T;

        int64_t size = 10000;
        this->RandomData(size);

        std::vector<T> &draws = this->draws_;
        std::vector<uint8_t> &valid_bytes = this->valid_bytes_;

        // first slug
        int64_t K = 1000;

        ASSERT_OK(this->builder_->append_values(draws.data(), K, valid_bytes.data()));

        std::shared_ptr<Array> out;
        FinishAndCheckPadding(this->builder_.get(), &out);
    }

    TYPED_TEST(TestPrimitiveBuilder, TestAppendValuesStdBool) {
        // ARROW-1383
        typedef typename TestFixture::CType T;

        int64_t size = 10000;
        this->RandomData(size);

        std::vector<T> &draws = this->draws_;

        std::vector<bool> is_valid;

        // first slug
        int64_t K = 1000;

        for (int64_t i = 0; i < K; ++i) {
            is_valid.push_back(this->valid_bytes_[i] != 0);
        }
        ASSERT_OK(this->builder_->append_values(draws.data(), K, is_valid));
        ASSERT_OK(this->builder_nn_->append_values(draws.data(), K));

        ASSERT_EQ(1000, this->builder_->length());
        ASSERT_EQ(1000, this->builder_->capacity());
        ASSERT_EQ(1000, this->builder_nn_->length());
        ASSERT_EQ(1000, this->builder_nn_->capacity());

        // append the next 9000
        is_valid.clear();
        std::vector<T> partial_draws;
        for (int64_t i = K; i < size; ++i) {
            partial_draws.push_back(draws[i]);
            is_valid.push_back(this->valid_bytes_[i] != 0);
        }

        ASSERT_OK(this->builder_->append_values(partial_draws, is_valid));
        ASSERT_OK(this->builder_nn_->append_values(partial_draws));

        ASSERT_EQ(size, this->builder_->length());
        ASSERT_GE(size, this->builder_->capacity());

        ASSERT_EQ(size, this->builder_nn_->length());
        ASSERT_GE(size, this->builder_->capacity());

        this->Check(this->builder_, true);
        this->Check(this->builder_nn_, false);
    }

    TYPED_TEST(TestPrimitiveBuilder, TestResize) {
        int64_t cap = kMinBuilderCapacity * 2;

        ASSERT_OK(this->builder_->Reserve(cap));
        ASSERT_EQ(cap, this->builder_->capacity());
    }

    TYPED_TEST(TestPrimitiveBuilder, TestReserve) {
        ASSERT_OK(this->builder_->Reserve(10));
        ASSERT_EQ(0, this->builder_->length());
        ASSERT_EQ(kMinBuilderCapacity, this->builder_->capacity());

        ASSERT_OK(this->builder_->Reserve(100));
        ASSERT_EQ(0, this->builder_->length());
        ASSERT_GE(100, this->builder_->capacity());
        ASSERT_OK(this->builder_->append_empty_values(100));
        ASSERT_EQ(100, this->builder_->length());
        ASSERT_GE(100, this->builder_->capacity());

        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, this->builder_->resize(1));
    }

    TEST(TestBooleanBuilder, AppendNullsAdvanceBuilder) {
        BooleanBuilder builder;

        std::vector<uint8_t> values = {1, 0, 0, 1};
        std::vector<uint8_t> is_valid = {1, 1, 0, 1};

        std::shared_ptr<Array> arr;
        ASSERT_OK(builder.append_values(values.data(), 2));
        ASSERT_OK(builder.append_nulls(1));
        ASSERT_OK(builder.append_values(values.data() + 3, 1));
        ASSERT_OK(builder.finish(&arr));

        ASSERT_EQ(1, arr->null_count());

        const auto &barr = static_cast<const BooleanArray &>(*arr);
        ASSERT_TRUE(barr.value(0));
        ASSERT_FALSE(barr.value(1));
        ASSERT_TRUE(barr.is_null(2));
        ASSERT_TRUE(barr.value(3));
    }

    TEST(TestBooleanBuilder, TestStdBoolVectorAppend) {
        BooleanBuilder builder;
        BooleanBuilder builder_nn;

        std::vector<bool> values, is_valid;

        const int length = 10000;
        random_is_valid(length, 0.5, &values);
        random_is_valid(length, 0.1, &is_valid);

        const int chunksize = 1000;
        for (int chunk = 0; chunk < length / chunksize; ++chunk) {
            std::vector<bool> chunk_values, chunk_is_valid;
            for (int i = chunk * chunksize; i < (chunk + 1) * chunksize; ++i) {
                chunk_values.push_back(values[i]);
                chunk_is_valid.push_back(is_valid[i]);
            }
            ASSERT_OK(builder.append_values(chunk_values, chunk_is_valid));
            ASSERT_OK(builder_nn.append_values(chunk_values));
        }

        std::shared_ptr<Array> result, result_nn;
        ASSERT_OK(builder.finish(&result));
        ASSERT_OK(builder_nn.finish(&result_nn));

        const auto &arr = turbo::checked_cast<const BooleanArray &>(*result);
        const auto &arr_nn = turbo::checked_cast<const BooleanArray &>(*result_nn);
        for (int i = 0; i < length; ++i) {
            if (is_valid[i]) {
                ASSERT_FALSE(arr.is_null(i));
                ASSERT_EQ(values[i], arr.value(i));
            } else {
                ASSERT_TRUE(arr.is_null(i));
            }
            ASSERT_EQ(values[i], arr_nn.value(i));
        }
    }

    template<typename TYPE>
    void CheckApproxEquals() {
        std::shared_ptr<Array> a, b;
        std::shared_ptr<DataType> type = TypeTraits<TYPE>::type_singleton();

        ArrayFromVector<TYPE>(type, {true, false}, {0.5, 1.0}, &a);
        ArrayFromVector<TYPE>(type, {true, false}, {0.5000001f, 2.0}, &b);
        ASSERT_TRUE(a->approx_equals(b));
        ASSERT_TRUE(b->approx_equals(a));
        for (bool nans_equal: {false, true}) {
            for (bool signed_zeros_equal: {false, true}) {
                auto opts =
                        EqualOptions().nans_equal(nans_equal).signed_zeros_equal(signed_zeros_equal);
                ASSERT_TRUE(a->approx_equals(b, opts));
                ASSERT_TRUE(b->approx_equals(a, opts));
            }
        }

        // Default tolerance too small
        ArrayFromVector<TYPE>(type, {true, false}, {0.5001f, 2.0}, &b);
        ASSERT_FALSE(a->approx_equals(b));
        ASSERT_FALSE(b->approx_equals(a));
        ASSERT_TRUE(a->approx_equals(b, EqualOptions().atol(1e-3)));
        ASSERT_TRUE(b->approx_equals(a, EqualOptions().atol(1e-3)));
        for (bool nans_equal: {false, true}) {
            for (bool signed_zeros_equal: {false, true}) {
                auto opts =
                        EqualOptions().nans_equal(nans_equal).signed_zeros_equal(signed_zeros_equal);
                ASSERT_FALSE(a->approx_equals(b, opts));
                ASSERT_FALSE(b->approx_equals(a, opts));
                ASSERT_TRUE(a->approx_equals(b, opts.atol(1e-3)));
                ASSERT_TRUE(b->approx_equals(a, opts.atol(1e-3)));
            }
        }

        // Values on other sides of 0
        ArrayFromVector<TYPE>(type, {true, false}, {-0.0001f, 1.0}, &a);
        ArrayFromVector<TYPE>(type, {true, false}, {0.0001f, 2.0}, &b);
        ASSERT_FALSE(a->approx_equals(b));
        ASSERT_FALSE(b->approx_equals(a));
        ASSERT_TRUE(a->approx_equals(b, EqualOptions().atol(1e-3)));
        ASSERT_TRUE(b->approx_equals(a, EqualOptions().atol(1e-3)));
        for (bool nans_equal: {false, true}) {
            for (bool signed_zeros_equal: {false, true}) {
                auto opts =
                        EqualOptions().nans_equal(nans_equal).signed_zeros_equal(signed_zeros_equal);
                ASSERT_FALSE(a->approx_equals(b, opts));
                ASSERT_FALSE(b->approx_equals(a, opts));
                ASSERT_TRUE(a->approx_equals(b, opts.atol(1e-3)));
                ASSERT_TRUE(b->approx_equals(a, opts.atol(1e-3)));
            }
        }

        // Mismatching validity
        ArrayFromVector<TYPE>(type, {true, false}, {0.5, 1.0}, &a);
        ArrayFromVector<TYPE>(type, {false, false}, {0.5, 1.0}, &b);
        ASSERT_FALSE(a->approx_equals(b));
        ASSERT_FALSE(b->approx_equals(a));
        ASSERT_FALSE(a->approx_equals(b, EqualOptions().nans_equal(true)));
        ASSERT_FALSE(b->approx_equals(a, EqualOptions().nans_equal(true)));

        ArrayFromVector<TYPE>(type, {false, true}, {0.5, 1.0}, &b);
        ASSERT_FALSE(a->approx_equals(b));
        ASSERT_FALSE(b->approx_equals(a));
        ASSERT_FALSE(a->approx_equals(b, EqualOptions().nans_equal(true)));
        ASSERT_FALSE(b->approx_equals(a, EqualOptions().nans_equal(true)));
    }

    template<typename TYPE>
    void CheckSliceApproxEquals() {
        using T = typename TYPE::c_type;

        const int64_t kSize = 50;
        std::vector<T> draws1;
        std::vector<T> draws2;

        const uint32_t kSeed = 0;
        random_real(kSize, kSeed, 0.0, 100.0, &draws1);
        random_real(kSize, kSeed + 1, 0.0, 100.0, &draws2);

        // Make the draws equal in the sliced segment, but unequal elsewhere (to
        // catch not using the slice offset)
        for (int64_t i = 10; i < 30; ++i) {
            draws2[i] = draws1[i];
        }

        std::vector<bool> is_valid;
        random_is_valid(kSize, 0.1, &is_valid);

        std::shared_ptr<Array> array1, array2;
        ArrayFromVector<TYPE, T>(is_valid, draws1, &array1);
        ArrayFromVector<TYPE, T>(is_valid, draws2, &array2);

        std::shared_ptr<Array> slice1 = array1->slice(10, 20);
        std::shared_ptr<Array> slice2 = array2->slice(10, 20);

        ASSERT_TRUE(slice1->approx_equals(slice2));
    }

    template<typename TYPE>
    void CheckFloatingNanEquality() {
        std::shared_ptr<Array> a, b;
        std::shared_ptr<DataType> type = TypeTraits<TYPE>::type_singleton();

        const auto nan_value = static_cast<typename TYPE::c_type>(NAN);

        // NaN in a null entry
        ArrayFromVector<TYPE>(type, {true, false}, {0.5, nan_value}, &a);
        ArrayFromVector<TYPE>(type, {true, false}, {0.5, nan_value}, &b);
        ASSERT_TRUE(a->equals(b));
        ASSERT_TRUE(b->equals(a));
        ASSERT_TRUE(a->approx_equals(b));
        ASSERT_TRUE(b->approx_equals(a));
        ASSERT_TRUE(a->range_equals(b, 0, 2, 0));
        ASSERT_TRUE(b->range_equals(a, 0, 2, 0));
        ASSERT_TRUE(a->range_equals(b, 1, 2, 1));
        ASSERT_TRUE(b->range_equals(a, 1, 2, 1));

        // NaN in a valid entry
        ArrayFromVector<TYPE>(type, {false, true}, {0.5, nan_value}, &a);
        ArrayFromVector<TYPE>(type, {false, true}, {0.5, nan_value}, &b);
        ASSERT_FALSE(a->equals(b));
        ASSERT_FALSE(b->equals(a));
        ASSERT_TRUE(a->equals(b, EqualOptions().nans_equal(true)));
        ASSERT_TRUE(b->equals(a, EqualOptions().nans_equal(true)));
        ASSERT_FALSE(a->approx_equals(b));
        ASSERT_FALSE(b->approx_equals(a));
        ASSERT_TRUE(a->approx_equals(b, EqualOptions().atol(1e-5).nans_equal(true)));
        ASSERT_TRUE(b->approx_equals(a, EqualOptions().atol(1e-5).nans_equal(true)));
        // NaN in tested range
        ASSERT_FALSE(a->range_equals(b, 0, 2, 0));
        ASSERT_FALSE(b->range_equals(a, 0, 2, 0));
        ASSERT_FALSE(a->range_equals(b, 1, 2, 1));
        ASSERT_FALSE(b->range_equals(a, 1, 2, 1));
        // NaN not in tested range
        ASSERT_TRUE(a->range_equals(b, 0, 1, 0));
        ASSERT_TRUE(b->range_equals(a, 0, 1, 0));

        // NaN != non-NaN
        ArrayFromVector<TYPE>(type, {false, true}, {0.5, nan_value}, &a);
        ArrayFromVector<TYPE>(type, {false, true}, {0.5, 0.0}, &b);
        ASSERT_FALSE(a->equals(b));
        ASSERT_FALSE(b->equals(a));
        ASSERT_FALSE(a->equals(b, EqualOptions().nans_equal(true)));
        ASSERT_FALSE(b->equals(a, EqualOptions().nans_equal(true)));
        ASSERT_FALSE(a->approx_equals(b));
        ASSERT_FALSE(b->approx_equals(a));
        ASSERT_FALSE(a->approx_equals(b, EqualOptions().atol(1e-5).nans_equal(true)));
        ASSERT_FALSE(b->approx_equals(a, EqualOptions().atol(1e-5).nans_equal(true)));
        // NaN in tested range
        ASSERT_FALSE(a->range_equals(b, 0, 2, 0));
        ASSERT_FALSE(b->range_equals(a, 0, 2, 0));
        ASSERT_FALSE(a->range_equals(b, 1, 2, 1));
        ASSERT_FALSE(b->range_equals(a, 1, 2, 1));
        // NaN not in tested range
        ASSERT_TRUE(a->range_equals(b, 0, 1, 0));
        ASSERT_TRUE(b->range_equals(a, 0, 1, 0));
    }

    template<typename TYPE>
    void CheckFloatingInfinityEquality() {
        std::shared_ptr<Array> a, b;
        std::shared_ptr<DataType> type = TypeTraits<TYPE>::type_singleton();

        const auto infinity = std::numeric_limits<typename TYPE::c_type>::infinity();

        for (auto nans_equal: {false, true}) {
            // Infinity in a null entry
            ArrayFromVector<TYPE>(type, {true, false}, {0.5, infinity}, &a);
            ArrayFromVector<TYPE>(type, {true, false}, {0.5, -infinity}, &b);
            ASSERT_TRUE(a->equals(b));
            ASSERT_TRUE(b->equals(a));
            ASSERT_TRUE(a->approx_equals(b, EqualOptions().atol(1e-5).nans_equal(nans_equal)));
            ASSERT_TRUE(b->approx_equals(a, EqualOptions().atol(1e-5).nans_equal(nans_equal)));
            ASSERT_TRUE(a->range_equals(b, 0, 2, 0));
            ASSERT_TRUE(b->range_equals(a, 0, 2, 0));
            ASSERT_TRUE(a->range_equals(b, 1, 2, 1));
            ASSERT_TRUE(b->range_equals(a, 1, 2, 1));

            // Infinity in a valid entry
            ArrayFromVector<TYPE>(type, {false, true}, {0.5, infinity}, &a);
            ArrayFromVector<TYPE>(type, {false, true}, {0.5, infinity}, &b);
            ASSERT_TRUE(a->equals(b));
            ASSERT_TRUE(b->equals(a));
            ASSERT_TRUE(a->approx_equals(b, EqualOptions().atol(1e-5).nans_equal(nans_equal)));
            ASSERT_TRUE(b->approx_equals(a, EqualOptions().atol(1e-5).nans_equal(nans_equal)));
            ASSERT_TRUE(a->approx_equals(b, EqualOptions().atol(1e-5).nans_equal(nans_equal)));
            ASSERT_TRUE(b->approx_equals(a, EqualOptions().atol(1e-5).nans_equal(nans_equal)));
            // Infinity in tested range
            ASSERT_TRUE(a->range_equals(b, 0, 2, 0));
            ASSERT_TRUE(b->range_equals(a, 0, 2, 0));
            ASSERT_TRUE(a->range_equals(b, 1, 2, 1));
            ASSERT_TRUE(b->range_equals(a, 1, 2, 1));
            // Infinity not in tested range
            ASSERT_TRUE(a->range_equals(b, 0, 1, 0));
            ASSERT_TRUE(b->range_equals(a, 0, 1, 0));

            // Infinity != non-infinity
            ArrayFromVector<TYPE>(type, {false, true}, {0.5, -infinity}, &a);
            ArrayFromVector<TYPE>(type, {false, true}, {0.5, 0.0}, &b);
            ASSERT_FALSE(a->equals(b));
            ASSERT_FALSE(b->equals(a));
            ASSERT_FALSE(a->approx_equals(b, EqualOptions().atol(1e-5).nans_equal(nans_equal)));
            ASSERT_FALSE(b->approx_equals(a));
            ASSERT_FALSE(a->approx_equals(b, EqualOptions().atol(1e-5).nans_equal(nans_equal)));
            ASSERT_FALSE(b->approx_equals(a, EqualOptions().atol(1e-5).nans_equal(nans_equal)));
            // Infinity != Negative infinity
            ArrayFromVector<TYPE>(type, {true, true}, {0.5, -infinity}, &a);
            ArrayFromVector<TYPE>(type, {true, true}, {0.5, infinity}, &b);
            ASSERT_FALSE(a->equals(b));
            ASSERT_FALSE(b->equals(a));
            ASSERT_FALSE(a->approx_equals(b));
            ASSERT_FALSE(b->approx_equals(a));
            ASSERT_FALSE(a->approx_equals(b, EqualOptions().atol(1e-5).nans_equal(nans_equal)));
            ASSERT_FALSE(b->approx_equals(a, EqualOptions().atol(1e-5).nans_equal(nans_equal)));
            // Infinity in tested range
            ASSERT_FALSE(a->range_equals(b, 0, 2, 0));
            ASSERT_FALSE(b->range_equals(a, 0, 2, 0));
            ASSERT_FALSE(a->range_equals(b, 1, 2, 1));
            ASSERT_FALSE(b->range_equals(a, 1, 2, 1));
            // Infinity not in tested range
            ASSERT_TRUE(a->range_equals(b, 0, 1, 0));
            ASSERT_TRUE(b->range_equals(a, 0, 1, 0));
        }
    }

    template<typename TYPE>
    void CheckFloatingZeroEquality() {
        std::shared_ptr<Array> a, b;
        std::shared_ptr<DataType> type = TypeTraits<TYPE>::type_singleton();

        ArrayFromVector<TYPE>(type, {true, false}, {0.0, 1.0}, &a);
        ArrayFromVector<TYPE>(type, {true, false}, {0.0, 1.0}, &b);
        ASSERT_TRUE(a->equals(b));
        ASSERT_TRUE(b->equals(a));
        for (auto nans_equal: {false, true}) {
            for (auto signed_zeros_equal: {false, true}) {
                auto opts =
                        EqualOptions().nans_equal(nans_equal).signed_zeros_equal(signed_zeros_equal);
                ASSERT_TRUE(a->equals(b, opts));
                ASSERT_TRUE(b->equals(a, opts));
                ASSERT_TRUE(a->range_equals(b, 0, 2, 0, opts));
                ASSERT_TRUE(b->range_equals(a, 0, 2, 0, opts));
                ASSERT_TRUE(a->range_equals(b, 1, 2, 1, opts));
                ASSERT_TRUE(b->range_equals(a, 1, 2, 1, opts));
            }
        }

        ArrayFromVector<TYPE>(type, {true, false}, {0.0, 1.0}, &a);
        ArrayFromVector<TYPE>(type, {true, false}, {-0.0, 1.0}, &b);
        for (auto nans_equal: {false, true}) {
            auto opts = EqualOptions().nans_equal(nans_equal);
            ASSERT_TRUE(a->equals(b, opts));
            ASSERT_TRUE(b->equals(a, opts));
            ASSERT_FALSE(a->equals(b, opts.signed_zeros_equal(false)));
            ASSERT_FALSE(b->equals(a, opts.signed_zeros_equal(false)));
            ASSERT_TRUE(a->range_equals(b, 0, 2, 0));
            ASSERT_TRUE(b->range_equals(a, 0, 2, 0));
            ASSERT_FALSE(a->range_equals(b, 0, 2, 0, opts.signed_zeros_equal(false)));
            ASSERT_FALSE(b->range_equals(a, 0, 2, 0, opts.signed_zeros_equal(false)));
            ASSERT_TRUE(a->range_equals(b, 1, 2, 1));
            ASSERT_TRUE(b->range_equals(a, 1, 2, 1));
            ASSERT_TRUE(a->range_equals(b, 1, 2, 1, opts.signed_zeros_equal(false)));
            ASSERT_TRUE(b->range_equals(a, 1, 2, 1, opts.signed_zeros_equal(false)));
        }
    }

    TEST(TestPrimitiveAdHoc, FloatingApproxEquals) {
        CheckApproxEquals<Fp32Type>();
        CheckApproxEquals<Fp64Type>();
    }

    TEST(TestPrimitiveAdHoc, FloatingSliceApproxEquals) {
        CheckSliceApproxEquals<Fp32Type>();
        CheckSliceApproxEquals<Fp64Type>();
    }

    TEST(TestPrimitiveAdHoc, FloatingNanEquality) {
        CheckFloatingNanEquality<Fp32Type>();
        CheckFloatingNanEquality<Fp64Type>();
    }

    TEST(TestPrimitiveAdHoc, FloatingInfinityEquality) {
        CheckFloatingInfinityEquality<Fp32Type>();
        CheckFloatingInfinityEquality<Fp64Type>();
    }

    TEST(TestPrimitiveAdHoc, FloatingZeroEquality) {
        CheckFloatingZeroEquality<Fp32Type>();
        CheckFloatingZeroEquality<Fp64Type>();
    }

// ----------------------------------------------------------------------
// FixedSizeBinary tests

    class TestFWBinaryArray : public ::testing::Test {
    public:
        void SetUp() {}

        void InitBuilder(int byte_width) {
            auto type = fixed_size_binary(byte_width);
            builder_.reset(new FixedSizeBinaryBuilder(type, default_memory_pool()));
        }

    protected:
        std::unique_ptr<FixedSizeBinaryBuilder> builder_;
    };

    TEST_F(TestFWBinaryArray, Builder) {
        int32_t byte_width = 10;
        int64_t length = 4096;

        int64_t nbytes = length * byte_width;

        std::vector<uint8_t> data(nbytes);
        random_bytes(nbytes, 0, data.data());

        std::vector<uint8_t> is_valid(length);
        random_null_bytes(length, 0.1, is_valid.data());

        const uint8_t *raw_data = data.data();

        std::shared_ptr<Array> result;

        auto CheckResult = [&length, &is_valid, &raw_data, &byte_width](const Array &result) {
            // Verify output
            const auto &fw_result = turbo::checked_cast<const FixedSizeBinaryArray &>(result);

            ASSERT_EQ(length, result.length());

            for (int64_t i = 0; i < result.length(); ++i) {
                if (is_valid[i]) {
                    ASSERT_EQ(0,
                              memcmp(raw_data + byte_width * i, fw_result.get_value(i), byte_width));
                } else {
                    ASSERT_TRUE(fw_result.is_null(i));
                }
            }
        };

        // Build using iterative API
        InitBuilder(byte_width);
        for (int64_t i = 0; i < length; ++i) {
            if (is_valid[i]) {
                ASSERT_OK(builder_->append(raw_data + byte_width * i));
            } else {
                ASSERT_OK(builder_->append_null());
            }
        }

        FinishAndCheckPadding(builder_.get(), &result);
        CheckResult(*result);

        // Build using batch API
        InitBuilder(byte_width);

        const uint8_t *raw_is_valid = is_valid.data();

        ASSERT_OK(builder_->append_values(raw_data, 50, raw_is_valid));
        ASSERT_OK(
                builder_->append_values(raw_data + 50 * byte_width, length - 50, raw_is_valid + 50));
        FinishAndCheckPadding(builder_.get(), &result);

        CheckResult(*result);

        // Build from std::string
        InitBuilder(byte_width);
        for (int64_t i = 0; i < length; ++i) {
            if (is_valid[i]) {
                ASSERT_OK(builder_->append(std::string(
                        reinterpret_cast<const char *>(raw_data + byte_width * i), byte_width)));
            } else {
                ASSERT_OK(builder_->append_null());
            }
        }

        ASSERT_OK(builder_->finish(&result));
        CheckResult(*result);
    }

    TEST_F(TestFWBinaryArray, EqualsRangeEquals) {
        // Check that we don't compare data in null slots

        auto type = fixed_size_binary(4);
        FixedSizeBinaryBuilder builder1(type);
        FixedSizeBinaryBuilder builder2(type);

        ASSERT_OK(builder1.append("foo1"));
        ASSERT_OK(builder1.append_null());

        ASSERT_OK(builder2.append("foo1"));
        ASSERT_OK(builder2.append("foo2"));

        std::shared_ptr<Array> array1, array2;
        ASSERT_OK(builder1.finish(&array1));
        ASSERT_OK(builder2.finish(&array2));

        const auto &a1 = turbo::checked_cast<const FixedSizeBinaryArray &>(*array1);
        const auto &a2 = turbo::checked_cast<const FixedSizeBinaryArray &>(*array2);

        FixedSizeBinaryArray equal1(type, 2, a1.values(), a1.null_bitmap(), 1);
        FixedSizeBinaryArray equal2(type, 2, a2.values(), a1.null_bitmap(), 1);

        ASSERT_TRUE(equal1.equals(equal2));
        ASSERT_TRUE(equal1.range_equals(equal2, 0, 2, 0));
    }

    TEST_F(TestFWBinaryArray, ZeroSize) {
        auto type = fixed_size_binary(0);
        FixedSizeBinaryBuilder builder(type);

        ASSERT_OK(builder.append(""));
        ASSERT_OK(builder.append(std::string()));
        ASSERT_OK(builder.append_null());
        ASSERT_OK(builder.append_null());
        ASSERT_OK(builder.append_null());

        std::shared_ptr<Array> array;
        ASSERT_OK(builder.finish(&array));

        const auto &fw_array = turbo::checked_cast<const FixedSizeBinaryArray &>(*array);

        // data is never allocated
        ASSERT_EQ(fw_array.values()->size(), 0);
        ASSERT_EQ(0, fw_array.byte_width());

        ASSERT_EQ(5, array->length());
        ASSERT_EQ(3, array->null_count());
    }

    TEST_F(TestFWBinaryArray, zero_padding) {
        auto type = fixed_size_binary(4);
        FixedSizeBinaryBuilder builder(type);

        ASSERT_OK(builder.append("foo1"));
        ASSERT_OK(builder.append_null());
        ASSERT_OK(builder.append("foo2"));
        ASSERT_OK(builder.append_null());
        ASSERT_OK(builder.append("foo3"));

        std::shared_ptr<Array> array;
        FinishAndCheckPadding(&builder, &array);
    }

    TEST_F(TestFWBinaryArray, slice) {
        auto type = fixed_size_binary(4);
        FixedSizeBinaryBuilder builder(type);

        std::vector<std::string> strings = {"foo1", "foo2", "foo3", "foo4", "foo5"};
        std::vector<uint8_t> is_null = {0, 1, 0, 0, 0};

        for (int i = 0; i < 5; ++i) {
            if (is_null[i]) {
                ASSERT_OK(builder.append_null());
            } else {
                ASSERT_OK(builder.append(strings[i]));
            }
        }

        std::shared_ptr<Array> array;
        ASSERT_OK(builder.finish(&array));

        std::shared_ptr<Array> slice, slice2;

        slice = array->slice(1);
        slice2 = array->slice(1);
        ASSERT_EQ(4, slice->length());

        ASSERT_TRUE(slice->equals(slice2));
        ASSERT_TRUE(array->range_equals(1, slice->length(), 0, slice));

        // Chained slices
        slice = array->slice(2);
        slice2 = array->slice(1)->slice(1);
        ASSERT_TRUE(slice->equals(slice2));

        slice = array->slice(1, 3);
        ASSERT_EQ(3, slice->length());

        slice2 = array->slice(1, 3);
        ASSERT_TRUE(slice->equals(slice2));
        ASSERT_TRUE(array->range_equals(1, 3, 0, slice));
    }

    TEST_F(TestFWBinaryArray, BuilderNulls) {
        auto type = fixed_size_binary(4);
        FixedSizeBinaryBuilder builder(type);

        for (int x = 0; x < 100; x++) {
            ASSERT_OK(builder.append_null());
        }
        ASSERT_OK(builder.append_nulls(500));

        std::shared_ptr<Array> array;
        ASSERT_OK(builder.finish(&array));

        for (auto buffer: array->data()->buffers) {
            for (int64_t i = 0; i < buffer->capacity(); i++) {
                // Validates current implementation, algorithms shouldn't rely on this
                ASSERT_EQ(0, *(buffer->data() + i)) << i;
            }
        }
    }

    struct FWBinaryAppender {
        turbo::Status VisitNull() {
            data.emplace_back("(null)");
            return turbo::OkStatus();
        }

        turbo::Status VisitValue(std::string_view v) {
            data.push_back(v);
            return turbo::OkStatus();
        }

        std::vector<std::string_view> data;
    };

    TEST_F(TestFWBinaryArray, ArraySpanVisitor) {
        auto type = fixed_size_binary(3);

        auto array = assert_array_from_json(type, R"(["abc", null, "def"])");
        FWBinaryAppender appender;
        ArraySpanVisitor<FixedSizeBinaryType> visitor;
        ASSERT_OK(visitor.Visit(*array->data(), &appender));
        ASSERT_THAT(appender.data, ::testing::ElementsAreArray({"abc", "(null)", "def"}));
        TURBO_UNUSED(visitor);  // Workaround weird MSVC warning
    }

    TEST_F(TestFWBinaryArray, ArraySpanVisitorSliced) {
        auto type = fixed_size_binary(3);

        auto array = assert_array_from_json(type, R"(["abc", null, "def", "ghi"])")->slice(1, 2);
        FWBinaryAppender appender;
        ArraySpanVisitor<FixedSizeBinaryType> visitor;
        ASSERT_OK(visitor.Visit(*array->data(), &appender));
        ASSERT_THAT(appender.data, ::testing::ElementsAreArray({"(null)", "def"}));
        TURBO_UNUSED(visitor);  // Workaround weird MSVC warning
    }

    TEST_F(TestFWBinaryArray, ArrayIndexOperator) {
        auto type = fixed_size_binary(3);
        auto arr = assert_array_from_json(type, R"(["abc", null, "def"])");
        auto fsba = turbo::checked_pointer_cast<FixedSizeBinaryArray>(arr);

        ASSERT_EQ("abc", (*fsba)[0].value());
        ASSERT_EQ(std::nullopt, (*fsba)[1]);
        ASSERT_EQ("def", (*fsba)[2].value());
    }

// ----------------------------------------------------------------------
// AdaptiveInt tests

    class TestAdaptiveIntBuilder : public TestBuilder {
    public:
        void SetUp() {
            TestBuilder::SetUp();
            builder_ = std::make_shared<AdaptiveIntBuilder>(pool_);
        }

        void EnsureValuesBufferNotNull(const ArrayData &data) {
            // The values buffer should be initialized
            ASSERT_EQ(2, data.buffers.size());
            ASSERT_NE(nullptr, data.buffers[1]);
            ASSERT_NE(nullptr, data.buffers[1]->data());
        }

        void Done() {
            FinishAndCheckPadding(builder_.get(), &result_);
            EnsureValuesBufferNotNull(*result_->data());
        }

        template<typename ExpectedType>
        void TestAppendValues() {
            using CType = typename TypeTraits<ExpectedType>::CType;
            auto type = TypeTraits<ExpectedType>::type_singleton();

            std::vector<int64_t> values(
                    {0, std::numeric_limits<CType>::min(), std::numeric_limits<CType>::max()});
            std::vector<CType> expected_values(
                    {0, std::numeric_limits<CType>::min(), std::numeric_limits<CType>::max()});
            ArrayFromVector<ExpectedType, CType>(expected_values, &expected_);

            SetUp();
            ASSERT_OK(builder_->append_values(values.data(), values.size()));
            AssertTypeEqual(*builder_->type(), *type);
            ASSERT_EQ(builder_->length(), static_cast<int64_t>(values.size()));
            Done();
            ASSERT_EQ(builder_->length(), 0);
            AssertArraysEqual(*expected_, *result_);

            // Reuse builder
            builder_->reset();
            AssertTypeEqual(*builder_->type(), *int8());
            ASSERT_OK(builder_->append_values(values.data(), values.size()));
            AssertTypeEqual(*builder_->type(), *type);
            ASSERT_EQ(builder_->length(), static_cast<int64_t>(values.size()));
            Done();
            ASSERT_EQ(builder_->length(), 0);
            AssertArraysEqual(*expected_, *result_);
        }

    protected:
        std::shared_ptr<AdaptiveIntBuilder> builder_;

        std::shared_ptr<Array> expected_;
        std::shared_ptr<Array> result_;
    };

    TEST_F(TestAdaptiveIntBuilder, TestInt8) {
        ASSERT_EQ(builder_->type()->id(), Type::INT8);
        ASSERT_EQ(builder_->length(), 0);
        ASSERT_OK(builder_->append(0));
        ASSERT_EQ(builder_->length(), 1);
        ASSERT_OK(builder_->append(127));
        ASSERT_EQ(builder_->length(), 2);
        ASSERT_OK(builder_->append(-128));
        ASSERT_EQ(builder_->length(), 3);

        Done();

        std::vector<int8_t> expected_values({0, 127, -128});
        ArrayFromVector<Int8Type, int8_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);
    }

    TEST_F(TestAdaptiveIntBuilder, TestInt16) {
        ASSERT_OK(builder_->append(0));
        ASSERT_EQ(builder_->type()->id(), Type::INT8);
        ASSERT_OK(builder_->append(128));
        ASSERT_EQ(builder_->type()->id(), Type::INT16);
        Done();

        std::vector<int16_t> expected_values({0, 128});
        ArrayFromVector<Int16Type, int16_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);

        SetUp();
        ASSERT_OK(builder_->append(-129));
        expected_values = {-129};
        Done();

        ArrayFromVector<Int16Type, int16_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);

        SetUp();
        ASSERT_OK(builder_->append(std::numeric_limits<int16_t>::max()));
        ASSERT_OK(builder_->append(std::numeric_limits<int16_t>::min()));
        expected_values = {std::numeric_limits<int16_t>::max(),
                           std::numeric_limits<int16_t>::min()};
        Done();

        ArrayFromVector<Int16Type, int16_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);
    }

    TEST_F(TestAdaptiveIntBuilder, TestInt16Nulls) {
        ASSERT_OK(builder_->append(0));
        ASSERT_EQ(builder_->type()->id(), Type::INT8);
        ASSERT_OK(builder_->append(128));
        ASSERT_EQ(builder_->type()->id(), Type::INT16);
        ASSERT_OK(builder_->append_null());
        ASSERT_EQ(builder_->type()->id(), Type::INT16);
        ASSERT_EQ(1, builder_->null_count());
        Done();

        std::vector<int16_t> expected_values({0, 128, 0});
        ArrayFromVector<Int16Type, int16_t>({1, 1, 0}, expected_values, &expected_);
        ASSERT_ARRAYS_EQUAL(*expected_, *result_);
    }

    TEST_F(TestAdaptiveIntBuilder, TestInt32) {
        ASSERT_OK(builder_->append(0));
        ASSERT_EQ(builder_->type()->id(), Type::INT8);
        ASSERT_OK(
                builder_->append(static_cast<int64_t>(std::numeric_limits<int16_t>::max()) + 1));
        ASSERT_EQ(builder_->type()->id(), Type::INT32);
        Done();

        std::vector<int32_t> expected_values(
                {0, static_cast<int32_t>(std::numeric_limits<int16_t>::max()) + 1});
        ArrayFromVector<Int32Type, int32_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);

        SetUp();
        ASSERT_OK(
                builder_->append(static_cast<int64_t>(std::numeric_limits<int16_t>::min()) - 1));
        expected_values = {static_cast<int32_t>(std::numeric_limits<int16_t>::min()) - 1};
        Done();

        ArrayFromVector<Int32Type, int32_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);

        SetUp();
        ASSERT_OK(builder_->append(std::numeric_limits<int32_t>::max()));
        ASSERT_OK(builder_->append(std::numeric_limits<int32_t>::min()));
        expected_values = {std::numeric_limits<int32_t>::max(),
                           std::numeric_limits<int32_t>::min()};
        Done();

        ArrayFromVector<Int32Type, int32_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);
    }

    TEST_F(TestAdaptiveIntBuilder, TestInt64) {
        ASSERT_OK(builder_->append(0));
        ASSERT_EQ(builder_->type()->id(), Type::INT8);
        ASSERT_OK(
                builder_->append(static_cast<int64_t>(std::numeric_limits<int32_t>::max()) + 1));
        ASSERT_EQ(builder_->type()->id(), Type::INT64);
        Done();

        std::vector<int64_t> expected_values(
                {0, static_cast<int64_t>(std::numeric_limits<int32_t>::max()) + 1});
        ArrayFromVector<Int64Type, int64_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);

        SetUp();
        ASSERT_OK(
                builder_->append(static_cast<int64_t>(std::numeric_limits<int32_t>::min()) - 1));
        expected_values = {static_cast<int64_t>(std::numeric_limits<int32_t>::min()) - 1};
        Done();

        ArrayFromVector<Int64Type, int64_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);

        SetUp();
        ASSERT_OK(builder_->append(std::numeric_limits<int64_t>::max()));
        ASSERT_OK(builder_->append(std::numeric_limits<int64_t>::min()));
        expected_values = {std::numeric_limits<int64_t>::max(),
                           std::numeric_limits<int64_t>::min()};
        Done();

        ArrayFromVector<Int64Type, int64_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);
    }

    TEST_F(TestAdaptiveIntBuilder, TestManyAppends) {
        // More than the builder's internal scratchpad size
        const int32_t n_values = 99999;
        std::vector<int32_t> expected_values(n_values);

        for (int32_t i = 0; i < n_values; ++i) {
            int32_t val = (i & 1) ? i : -i;
            expected_values[i] = val;
            ASSERT_OK(builder_->append(val));
            ASSERT_EQ(builder_->length(), i + 1);
        }
        ASSERT_EQ(builder_->type()->id(), Type::INT32);
        Done();

        ArrayFromVector<Int32Type, int32_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);
    }

    TEST_F(TestAdaptiveIntBuilder, TestAppendValues) {
        this->template TestAppendValues<Int64Type>();
        this->template TestAppendValues<Int32Type>();
        this->template TestAppendValues<Int16Type>();
        this->template TestAppendValues<Int8Type>();
    }

    TEST_F(TestAdaptiveIntBuilder, TestAssertZeroPadded) {
        std::vector<int64_t> values(
                {0, static_cast<int64_t>(std::numeric_limits<int32_t>::max()) + 1});
        ASSERT_OK(builder_->append_values(values.data(), values.size()));
        Done();
    }

    TEST_F(TestAdaptiveIntBuilder, TestAppendNull) {
        int64_t size = 1000;
        ASSERT_OK(builder_->append(127));
        ASSERT_EQ(0, builder_->null_count());
        for (unsigned index = 1; index < size - 1; ++index) {
            ASSERT_OK(builder_->append_null());
            ASSERT_EQ(index, builder_->null_count());
        }
        ASSERT_OK(builder_->append(-128));
        ASSERT_EQ(size - 2, builder_->null_count());

        Done();

        std::vector<bool> expected_valid(size, false);
        expected_valid[0] = true;
        expected_valid[size - 1] = true;
        std::vector<int8_t> expected_values(size);
        expected_values[0] = 127;
        expected_values[size - 1] = -128;
        std::shared_ptr<Array> expected;
        ArrayFromVector<Int8Type, int8_t>(expected_valid, expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);
    }

    TEST_F(TestAdaptiveIntBuilder, TestAppendNulls) {
        constexpr int64_t size = 10;
        ASSERT_EQ(0, builder_->null_count());
        ASSERT_OK(builder_->append_nulls(0));
        ASSERT_OK(builder_->append_nulls(size));
        ASSERT_OK(builder_->append_nulls(0));
        ASSERT_EQ(size, builder_->null_count());

        Done();

        for (unsigned index = 0; index < size; ++index) {
            ASSERT_FALSE(result_->is_valid(index));
        }
    }

    TEST_F(TestAdaptiveIntBuilder, TestAppendEmptyValue) {
        ASSERT_OK(builder_->append_empty_values(0));
        ASSERT_OK(builder_->append_nulls(2));
        ASSERT_OK(builder_->append_empty_value());
        ASSERT_OK(builder_->append(42));
        ASSERT_OK(builder_->append_empty_values(2));
        Done();

        ASSERT_OK(result_->validate_full());
        // NOTE: The fact that we get 0 is really an implementation detail
        AssertArraysEqual(*result_, *assert_array_from_json(int8(), "[null, null, 0, 42, 0, 0]"));
    }

    TEST_F(TestAdaptiveIntBuilder, Empty) { Done(); }

    TEST(TestAdaptiveIntBuilderWithStartIntSize, TestReset) {
        auto builder = std::make_shared<AdaptiveIntBuilder>(
                static_cast<uint8_t>(sizeof(int16_t)), default_memory_pool());
        AssertTypeEqual(*int16(), *builder->type());

        ASSERT_OK(
                builder->append(static_cast<int64_t>(std::numeric_limits<int16_t>::max()) + 1));
        AssertTypeEqual(*int32(), *builder->type());

        builder->reset();
        AssertTypeEqual(*int16(), *builder->type());
    }

    class TestAdaptiveUIntBuilder : public TestBuilder {
    public:
        void SetUp() {
            TestBuilder::SetUp();
            builder_ = std::make_shared<AdaptiveUIntBuilder>(pool_);
        }

        void Done() { FinishAndCheckPadding(builder_.get(), &result_); }

        template<typename ExpectedType>
        void TestAppendValues() {
            using CType = typename TypeTraits<ExpectedType>::CType;
            auto type = TypeTraits<ExpectedType>::type_singleton();

            std::vector<uint64_t> values(
                    {0, std::numeric_limits<CType>::min(), std::numeric_limits<CType>::max()});
            std::vector<CType> expected_values(
                    {0, std::numeric_limits<CType>::min(), std::numeric_limits<CType>::max()});
            ArrayFromVector<ExpectedType, CType>(expected_values, &expected_);

            SetUp();
            ASSERT_OK(builder_->append_values(values.data(), values.size()));
            AssertTypeEqual(*builder_->type(), *type);
            ASSERT_EQ(builder_->length(), static_cast<int64_t>(values.size()));
            Done();
            ASSERT_EQ(builder_->length(), 0);
            AssertArraysEqual(*expected_, *result_);

            // Reuse builder
            builder_->reset();
            AssertTypeEqual(*builder_->type(), *uint8());
            ASSERT_OK(builder_->append_values(values.data(), values.size()));
            AssertTypeEqual(*builder_->type(), *type);
            ASSERT_EQ(builder_->length(), static_cast<int64_t>(values.size()));
            Done();
            ASSERT_EQ(builder_->length(), 0);
            AssertArraysEqual(*expected_, *result_);
        }

    protected:
        std::shared_ptr<AdaptiveUIntBuilder> builder_;

        std::shared_ptr<Array> expected_;
        std::shared_ptr<Array> result_;
    };

    TEST_F(TestAdaptiveUIntBuilder, TestUInt8) {
        ASSERT_EQ(builder_->length(), 0);
        ASSERT_OK(builder_->append(0));
        ASSERT_EQ(builder_->length(), 1);
        ASSERT_OK(builder_->append(255));
        ASSERT_EQ(builder_->length(), 2);

        Done();

        std::vector<uint8_t> expected_values({0, 255});
        ArrayFromVector<UInt8Type, uint8_t>(expected_values, &expected_);
        ASSERT_TRUE(expected_->equals(result_));
    }

    TEST_F(TestAdaptiveUIntBuilder, TestUInt16) {
        ASSERT_OK(builder_->append(0));
        ASSERT_EQ(builder_->type()->id(), Type::UINT8);
        ASSERT_OK(builder_->append(256));
        ASSERT_EQ(builder_->type()->id(), Type::UINT16);
        Done();

        std::vector<uint16_t> expected_values({0, 256});
        ArrayFromVector<UInt16Type, uint16_t>(expected_values, &expected_);
        ASSERT_ARRAYS_EQUAL(*expected_, *result_);

        SetUp();
        ASSERT_OK(builder_->append(std::numeric_limits<uint16_t>::max()));
        expected_values = {std::numeric_limits<uint16_t>::max()};
        Done();

        ArrayFromVector<UInt16Type, uint16_t>(expected_values, &expected_);
        ASSERT_TRUE(expected_->equals(result_));
    }

    TEST_F(TestAdaptiveUIntBuilder, TestUInt16Nulls) {
        ASSERT_OK(builder_->append(0));
        ASSERT_EQ(builder_->type()->id(), Type::UINT8);
        ASSERT_OK(builder_->append(256));
        ASSERT_EQ(builder_->type()->id(), Type::UINT16);
        ASSERT_OK(builder_->append_null());
        ASSERT_EQ(builder_->type()->id(), Type::UINT16);
        ASSERT_EQ(1, builder_->null_count());
        Done();

        std::vector<uint16_t> expected_values({0, 256, 0});
        ArrayFromVector<UInt16Type, uint16_t>({1, 1, 0}, expected_values, &expected_);
        ASSERT_ARRAYS_EQUAL(*expected_, *result_);
    }

    TEST_F(TestAdaptiveUIntBuilder, TestUInt32) {
        ASSERT_OK(builder_->append(0));
        ASSERT_EQ(builder_->type()->id(), Type::UINT8);
        ASSERT_OK(
                builder_->append(static_cast<uint64_t>(std::numeric_limits<uint16_t>::max()) + 1));
        ASSERT_EQ(builder_->type()->id(), Type::UINT32);
        Done();

        std::vector<uint32_t> expected_values(
                {0, static_cast<uint32_t>(std::numeric_limits<uint16_t>::max()) + 1});
        ArrayFromVector<UInt32Type, uint32_t>(expected_values, &expected_);
        ASSERT_TRUE(expected_->equals(result_));

        SetUp();
        ASSERT_OK(builder_->append(std::numeric_limits<uint32_t>::max()));
        expected_values = {std::numeric_limits<uint32_t>::max()};
        Done();

        ArrayFromVector<UInt32Type, uint32_t>(expected_values, &expected_);
        ASSERT_TRUE(expected_->equals(result_));
    }

    TEST_F(TestAdaptiveUIntBuilder, TestUInt64) {
        ASSERT_OK(builder_->append(0));
        ASSERT_EQ(builder_->type()->id(), Type::UINT8);
        ASSERT_OK(
                builder_->append(static_cast<uint64_t>(std::numeric_limits<uint32_t>::max()) + 1));
        ASSERT_EQ(builder_->type()->id(), Type::UINT64);
        Done();

        std::vector<uint64_t> expected_values(
                {0, static_cast<uint64_t>(std::numeric_limits<uint32_t>::max()) + 1});
        ArrayFromVector<UInt64Type, uint64_t>(expected_values, &expected_);
        ASSERT_TRUE(expected_->equals(result_));

        SetUp();
        ASSERT_OK(builder_->append(std::numeric_limits<uint64_t>::max()));
        expected_values = {std::numeric_limits<uint64_t>::max()};
        Done();

        ArrayFromVector<UInt64Type, uint64_t>(expected_values, &expected_);
        ASSERT_TRUE(expected_->equals(result_));
    }

    TEST_F(TestAdaptiveUIntBuilder, TestManyAppends) {
        // More than the builder's internal scratchpad size
        const int32_t n_values = 99999;
        std::vector<uint32_t> expected_values(n_values);

        for (int32_t i = 0; i < n_values; ++i) {
            auto val = static_cast<uint32_t>(i);
            expected_values[i] = val;
            ASSERT_OK(builder_->append(val));
            ASSERT_EQ(builder_->length(), i + 1);
        }
        ASSERT_EQ(builder_->type()->id(), Type::UINT32);
        Done();

        ArrayFromVector<UInt32Type, uint32_t>(expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);
    }

    TEST_F(TestAdaptiveUIntBuilder, TestAppendValues) {
        this->template TestAppendValues<UInt64Type>();
        this->template TestAppendValues<UInt32Type>();
        this->template TestAppendValues<UInt16Type>();
        this->template TestAppendValues<UInt8Type>();
    }

    TEST_F(TestAdaptiveUIntBuilder, TestAssertZeroPadded) {
        std::vector<uint64_t> values(
                {0, static_cast<uint64_t>(std::numeric_limits<uint32_t>::max()) + 1});
        ASSERT_OK(builder_->append_values(values.data(), values.size()));
        Done();
    }

    TEST_F(TestAdaptiveUIntBuilder, TestAppendNull) {
        int64_t size = 1000;
        ASSERT_OK(builder_->append(254));
        for (unsigned index = 1; index < size - 1; ++index) {
            ASSERT_OK(builder_->append_null());
            ASSERT_EQ(index, builder_->null_count());
        }
        ASSERT_OK(builder_->append(255));

        Done();

        std::vector<bool> expected_valid(size, false);
        expected_valid[0] = true;
        expected_valid[size - 1] = true;
        std::vector<uint8_t> expected_values(size);
        expected_values[0] = 254;
        expected_values[size - 1] = 255;
        std::shared_ptr<Array> expected;
        ArrayFromVector<UInt8Type, uint8_t>(expected_valid, expected_values, &expected_);
        AssertArraysEqual(*expected_, *result_);
    }

    TEST_F(TestAdaptiveUIntBuilder, TestAppendNulls) {
        constexpr int64_t size = 10;
        ASSERT_OK(builder_->append_nulls(0));
        ASSERT_OK(builder_->append_nulls(size));
        ASSERT_OK(builder_->append_nulls(0));
        ASSERT_EQ(size, builder_->null_count());

        Done();

        for (unsigned index = 0; index < size; ++index) {
            ASSERT_FALSE(result_->is_valid(index));
        }
    }

    TEST_F(TestAdaptiveUIntBuilder, TestAppendEmptyValue) {
        ASSERT_OK(builder_->append_empty_values(0));
        ASSERT_OK(builder_->append_nulls(2));
        ASSERT_OK(builder_->append_empty_value());
        ASSERT_OK(builder_->append(42));
        ASSERT_OK(builder_->append_empty_values(2));
        Done();

        ASSERT_OK(result_->validate_full());
        // NOTE: The fact that we get 0 is really an implementation detail
        AssertArraysEqual(*result_, *assert_array_from_json(uint8(), "[null, null, 0, 42, 0, 0]"));
    }

    TEST(TestAdaptiveUIntBuilderWithStartIntSize, TestReset) {
        auto builder = std::make_shared<AdaptiveUIntBuilder>(
                static_cast<uint8_t>(sizeof(uint16_t)), default_memory_pool());
        AssertTypeEqual(uint16(), builder->type());

        ASSERT_OK(
                builder->append(static_cast<uint64_t>(std::numeric_limits<uint16_t>::max()) + 1));
        AssertTypeEqual(uint32(), builder->type());

        builder->reset();
        AssertTypeEqual(uint16(), builder->type());
    }

// ----------------------------------------------------------------------
// Test Decimal arrays

    template<typename TYPE>
    class DecimalTest : public ::testing::TestWithParam<int> {
    public:
        using DecimalBuilder = typename TypeTraits<TYPE>::BuilderType;
        using DecimalValue = typename TypeTraits<TYPE>::ScalarType::value_type;
        using DecimalArray = typename TypeTraits<TYPE>::ArrayType;
        using DecimalVector = std::vector<DecimalValue>;

        DecimalTest() {}

        template<size_t BYTE_WIDTH = 16>
        void MakeData(const DecimalVector &input, std::vector<uint8_t> *out) const {
            out->reserve(input.size() * BYTE_WIDTH);

            for (const auto &value: input) {
                auto bytes = value.ToBytes();
                out->insert(out->end(), bytes.cbegin(), bytes.cend());
            }
        }

        template<size_t BYTE_WIDTH = 16>
        std::shared_ptr<Array> TestCreate(int32_t precision, const DecimalVector &draw,
                                          const std::vector<uint8_t> &valid_bytes,
                                          int64_t offset) const {
            auto type = std::make_shared<TYPE>(precision, 4);
            auto builder = std::make_shared<DecimalBuilder>(type);

            const size_t size = draw.size();

            NEBULA_EXPECT_OK(builder->Reserve(size));

            for (size_t i = 0; i < size; ++i) {
                if (valid_bytes[i]) {
                    NEBULA_EXPECT_OK(builder->append(draw[i]));
                } else {
                    NEBULA_EXPECT_OK(builder->append_null());
                }
            }

            std::shared_ptr<Array> out;
            FinishAndCheckPadding(builder.get(), &out);
            EXPECT_EQ(builder->length(), 0);

            std::vector<uint8_t> raw_bytes;

            raw_bytes.reserve(size * BYTE_WIDTH);
            MakeData<BYTE_WIDTH>(draw, &raw_bytes);

            auto expected_data = std::make_shared<Buffer>(raw_bytes.data(), BYTE_WIDTH);
            std::shared_ptr<Buffer> expected_null_bitmap;
            EXPECT_OK_AND_ASSIGN(expected_null_bitmap, internal::BytesToBits(valid_bytes));

            int64_t expected_null_count = CountNulls(valid_bytes);
            auto expected = std::make_shared<DecimalArray>(
                    type, size, expected_data, expected_null_bitmap, expected_null_count);

            std::shared_ptr<Array> lhs = out->slice(offset);
            std::shared_ptr<Array> rhs = expected->slice(offset);
            ASSERT_ARRAYS_EQUAL(*rhs, *lhs);

            return out;
        }
    };

    using Decimal128Test = DecimalTest<Decimal128Type>;

    TEST_P(Decimal128Test, NoNulls) {
        int32_t precision = GetParam();
        std::vector<Decimal128> draw = {Decimal128(1), Decimal128(-2), Decimal128(2389),
                                        Decimal128(4), Decimal128(-12348)};
        std::vector<uint8_t> valid_bytes = {true, true, true, true, true};
        this->TestCreate(precision, draw, valid_bytes, 0);
        this->TestCreate(precision, draw, valid_bytes, 2);
    }

    TEST_P(Decimal128Test, WithNulls) {
        int32_t precision = GetParam();
        std::vector<Decimal128> draw = {Decimal128(1), Decimal128(2), Decimal128(-1),
                                        Decimal128(4), Decimal128(-1), Decimal128(1),
                                        Decimal128(2)};
        Decimal128 big;
        ASSERT_OK_AND_ASSIGN(big, Decimal128::from_string("230342903942.234234"));
        draw.push_back(big);

        Decimal128 big_negative;
        ASSERT_OK_AND_ASSIGN(big_negative, Decimal128::from_string("-23049302932.235234"));
        draw.push_back(big_negative);

        std::vector<uint8_t> valid_bytes = {true, true, false, true, false,
                                            true, true, true, true};
        this->TestCreate(precision, draw, valid_bytes, 0);
        this->TestCreate(precision, draw, valid_bytes, 2);
    }

    TEST_P(Decimal128Test, validate_full) {
        int32_t precision = GetParam();
        std::vector<Decimal128> draw;
        Decimal128 val = Decimal128::GetMaxValue(precision) + 1;

        draw = {Decimal128(), val};
        auto arr = this->TestCreate(precision, draw, {true, false}, 0);
        ASSERT_OK(arr->validate_full());

        draw = {val, Decimal128()};
        arr = this->TestCreate(precision, draw, {true, false}, 0);
        EXPECT_RAISES_WITH_MESSAGE_THAT(
                turbo::StatusCode::kInvalidArgument, ::testing::HasSubstr("does not fit in precision of"),
                arr->validate_full());
    }

    INSTANTIATE_TEST_SUITE_P(Decimal128Test, Decimal128Test, ::testing::Range(1, 38));

    using Decimal256Test = DecimalTest<Decimal256Type>;

    TEST_P(Decimal256Test, NoNulls) {
        int32_t precision = GetParam();
        std::vector<Decimal256> draw = {Decimal256(1), Decimal256(-2), Decimal256(2389),
                                        Decimal256(4), Decimal256(-12348)};
        std::vector<uint8_t> valid_bytes = {true, true, true, true, true};
        this->TestCreate(precision, draw, valid_bytes, 0);
        this->TestCreate(precision, draw, valid_bytes, 2);
    }

    TEST_P(Decimal256Test, WithNulls) {
        int32_t precision = GetParam();
        std::vector<Decimal256> draw = {Decimal256(1), Decimal256(2), Decimal256(-1),
                                        Decimal256(4), Decimal256(-1), Decimal256(1),
                                        Decimal256(2)};
        Decimal256 big;  // (pow(2, 255) - 1) / pow(10, 38)
        ASSERT_OK_AND_ASSIGN(big,
                             Decimal256::from_string("578960446186580977117854925043439539266."
                                                    "34992332820282019728792003956564819967"));
        draw.push_back(big);

        Decimal256 big_negative;  // -pow(2, 255) / pow(10, 38)
        ASSERT_OK_AND_ASSIGN(big_negative,
                             Decimal256::from_string("-578960446186580977117854925043439539266."
                                                    "34992332820282019728792003956564819968"));
        draw.push_back(big_negative);

        std::vector<uint8_t> valid_bytes = {true, true, false, true, false,
                                            true, true, true, true};
        this->TestCreate(precision, draw, valid_bytes, 0);
        this->TestCreate(precision, draw, valid_bytes, 2);
    }

    TEST_P(Decimal256Test, validate_full) {
        int32_t precision = GetParam();
        std::vector<Decimal256> draw;
        Decimal256 val = Decimal256::GetMaxValue(precision) + 1;

        draw = {Decimal256(), val};
        auto arr = this->TestCreate(precision, draw, {true, false}, 0);
        ASSERT_OK(arr->validate_full());

        draw = {val, Decimal256()};
        arr = this->TestCreate(precision, draw, {true, false}, 0);
        EXPECT_RAISES_WITH_MESSAGE_THAT(
                turbo::StatusCode::kInvalidArgument, ::testing::HasSubstr("does not fit in precision of"),
                arr->validate_full());
    }

    INSTANTIATE_TEST_SUITE_P(Decimal256Test, Decimal256Test,
                             ::testing::Values(1, 2, 5, 10, 38, 39, 40, 75, 76));

// ----------------------------------------------------------------------
// Test rechunking

    TEST(TestRechunkArraysConsistently, Trivial) {
        std::vector<ArrayVector> groups, rechunked;
        rechunked = internal::RechunkArraysConsistently(groups);
        ASSERT_EQ(rechunked.size(), 0);

        std::shared_ptr<Array> a1, a2, b1;
        ArrayFromVector<Int16Type, int16_t>({}, &a1);
        ArrayFromVector<Int16Type, int16_t>({}, &a2);
        ArrayFromVector<Int32Type, int32_t>({}, &b1);

        groups = {{a1, a2},
                  {},
                  {b1}};
        rechunked = internal::RechunkArraysConsistently(groups);
        ASSERT_EQ(rechunked.size(), 3);

        for (auto &arrvec: rechunked) {
            for (auto &arr: arrvec) {
                AssertZeroPadded(*arr);
                TestInitialized(*arr);
            }
        }
    }

    TEST(TestRechunkArraysConsistently, Plain) {
        std::shared_ptr<Array> expected;
        std::shared_ptr<Array> a1, a2, a3, b1, b2, b3, b4;
        ArrayFromVector<Int16Type, int16_t>({1, 2, 3}, &a1);
        ArrayFromVector<Int16Type, int16_t>({4, 5}, &a2);
        ArrayFromVector<Int16Type, int16_t>({6, 7, 8, 9}, &a3);

        ArrayFromVector<Int32Type, int32_t>({41, 42}, &b1);
        ArrayFromVector<Int32Type, int32_t>({43, 44, 45}, &b2);
        ArrayFromVector<Int32Type, int32_t>({46, 47}, &b3);
        ArrayFromVector<Int32Type, int32_t>({48, 49}, &b4);

        ArrayVector a{a1, a2, a3};
        ArrayVector b{b1, b2, b3, b4};

        std::vector<ArrayVector> groups{a, b}, rechunked;
        rechunked = internal::RechunkArraysConsistently(groups);
        ASSERT_EQ(rechunked.size(), 2);
        auto ra = rechunked[0];
        auto rb = rechunked[1];

        ASSERT_EQ(ra.size(), 5);
        ArrayFromVector<Int16Type, int16_t>({1, 2}, &expected);
        ASSERT_ARRAYS_EQUAL(*ra[0], *expected);
        ArrayFromVector<Int16Type, int16_t>({3}, &expected);
        ASSERT_ARRAYS_EQUAL(*ra[1], *expected);
        ArrayFromVector<Int16Type, int16_t>({4, 5}, &expected);
        ASSERT_ARRAYS_EQUAL(*ra[2], *expected);
        ArrayFromVector<Int16Type, int16_t>({6, 7}, &expected);
        ASSERT_ARRAYS_EQUAL(*ra[3], *expected);
        ArrayFromVector<Int16Type, int16_t>({8, 9}, &expected);
        ASSERT_ARRAYS_EQUAL(*ra[4], *expected);

        ASSERT_EQ(rb.size(), 5);
        ArrayFromVector<Int32Type, int32_t>({41, 42}, &expected);
        ASSERT_ARRAYS_EQUAL(*rb[0], *expected);
        ArrayFromVector<Int32Type, int32_t>({43}, &expected);
        ASSERT_ARRAYS_EQUAL(*rb[1], *expected);
        ArrayFromVector<Int32Type, int32_t>({44, 45}, &expected);
        ASSERT_ARRAYS_EQUAL(*rb[2], *expected);
        ArrayFromVector<Int32Type, int32_t>({46, 47}, &expected);
        ASSERT_ARRAYS_EQUAL(*rb[3], *expected);
        ArrayFromVector<Int32Type, int32_t>({48, 49}, &expected);
        ASSERT_ARRAYS_EQUAL(*rb[4], *expected);

        for (auto &arrvec: rechunked) {
            for (auto &arr: arrvec) {
                AssertZeroPadded(*arr);
                TestInitialized(*arr);
            }
        }
    }

// ----------------------------------------------------------------------
// Test SwapEndianArrayData

/// \brief Indicate if fields are equals.
///
/// \param[in] target ArrayData to be converted and tested
/// \param[in] expected result ArrayData
    void AssertArrayDataEqualsWithSwapEndian(const std::shared_ptr<ArrayData> &target,
                                             const std::shared_ptr<ArrayData> &expected) {
        auto swap_array = make_array(*::nebula::internal::SwapEndianArrayData(target));
        auto expected_array = make_array(expected);
        ASSERT_ARRAYS_EQUAL(*swap_array, *expected_array);
        ASSERT_OK(swap_array->validate_full());
    }

    TEST(TestSwapEndianArrayData, PrimitiveType) {
        auto null_buffer = Buffer::from_string("\xff");
        auto data_int_buffer = Buffer::from_string("01234567");

        auto data = ArrayData::create(null(), 0, {nullptr}, 0);
        auto expected_data = data;
        AssertArrayDataEqualsWithSwapEndian(data, expected_data);

        data = ArrayData::create(boolean(), 8, {null_buffer, data_int_buffer}, 0);
        expected_data = data;
        AssertArrayDataEqualsWithSwapEndian(data, expected_data);

        data = ArrayData::create(int8(), 8, {null_buffer, data_int_buffer}, 0);
        expected_data = data;
        AssertArrayDataEqualsWithSwapEndian(data, expected_data);

        data = ArrayData::create(uint16(), 4, {null_buffer, data_int_buffer}, 0);
        auto data_int16_buffer = Buffer::from_string("10325476");
        expected_data = ArrayData::create(uint16(), 4, {null_buffer, data_int16_buffer}, 0);
        AssertArrayDataEqualsWithSwapEndian(data, expected_data);

        data = ArrayData::create(int32(), 2, {null_buffer, data_int_buffer}, 0);
        auto data_int32_buffer = Buffer::from_string("32107654");
        expected_data = ArrayData::create(int32(), 2, {null_buffer, data_int32_buffer}, 0);
        AssertArrayDataEqualsWithSwapEndian(data, expected_data);

        data = ArrayData::create(uint64(), 1, {null_buffer, data_int_buffer}, 0);
        auto data_int64_buffer = Buffer::from_string("76543210");
        expected_data = ArrayData::create(uint64(), 1, {null_buffer, data_int64_buffer}, 0);
        AssertArrayDataEqualsWithSwapEndian(data, expected_data);

        auto data_16byte_buffer = Buffer::from_string(
                "\x01"
                "123456789abcde\x01");
        data = ArrayData::create(decimal128(38, 10), 1, {null_buffer, data_16byte_buffer});
        auto data_decimal128_buffer = Buffer::from_string(
                "\x01"
                "edcba987654321\x01");
        expected_data =
                ArrayData::create(decimal128(38, 10), 1, {null_buffer, data_decimal128_buffer}, 0);
        AssertArrayDataEqualsWithSwapEndian(data, expected_data);

        auto data_32byte_buffer = Buffer::from_string(
                "\x01"
                "123456789abcdef123456789ABCDEF\x01");
        data = ArrayData::create(decimal256(76, 20), 1, {null_buffer, data_32byte_buffer});
        auto data_decimal256_buffer = Buffer::from_string(
                "\x01"
                "FEDCBA987654321fedcba987654321\x01");
        expected_data =
                ArrayData::create(decimal256(76, 20), 1, {null_buffer, data_decimal256_buffer}, 0);
        AssertArrayDataEqualsWithSwapEndian(data, expected_data);

        auto data_float_buffer = Buffer::from_string("01200560");
        data = ArrayData::create(float32(), 2, {null_buffer, data_float_buffer}, 0);
        auto data_float32_buffer = Buffer::from_string("02100650");
        expected_data = ArrayData::create(float32(), 2, {null_buffer, data_float32_buffer}, 0);
        AssertArrayDataEqualsWithSwapEndian(data, expected_data);

        data = ArrayData::create(float64(), 1, {null_buffer, data_float_buffer});
        auto data_float64_buffer = Buffer::from_string("06500210");
        expected_data = ArrayData::create(float64(), 1, {null_buffer, data_float64_buffer}, 0);
        AssertArrayDataEqualsWithSwapEndian(data, expected_data);

        // With offset > 0
        data =
                ArrayData::create(int64(), 1, {null_buffer, data_int_buffer}, kUnknownNullCount, 1);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, ::nebula::internal::SwapEndianArrayData(data));
    }

    std::shared_ptr<ArrayData> ReplaceBuffers(const std::shared_ptr<ArrayData> &data,
                                              const int32_t buffer_index,
                                              const std::vector<uint8_t> &buffer_data) {
        const auto test_data = data->copy();
        test_data->buffers[buffer_index] =
                std::make_shared<Buffer>(buffer_data.data(), buffer_data.size());
        return test_data;
    }

    std::shared_ptr<ArrayData> ReplaceBuffersInChild(const std::shared_ptr<ArrayData> &data,
                                                     const int32_t child_index,
                                                     const std::vector<uint8_t> &child_data) {
        const auto test_data = data->copy();
        // assume updating only buffer[1] in child_data
        auto child_array_data = test_data->child_data[child_index]->copy();
        child_array_data->buffers[1] =
                std::make_shared<Buffer>(child_data.data(), child_data.size());
        test_data->child_data[child_index] = child_array_data;
        return test_data;
    }

    std::shared_ptr<ArrayData> ReplaceBuffersInDictionary(
            const std::shared_ptr<ArrayData> &data, const int32_t buffer_index,
            const std::vector<uint8_t> &buffer_data) {
        const auto test_data = data->copy();
        auto dict_array_data = test_data->dictionary->copy();
        dict_array_data->buffers[buffer_index] =
                std::make_shared<Buffer>(buffer_data.data(), buffer_data.size());
        test_data->dictionary = dict_array_data;
        return test_data;
    }

    TEST(TestSwapEndianArrayData, BinaryType) {
        auto array = assert_array_from_json(binary(), R"(["0123", null, "45"])");
        const std::vector<uint8_t> offset1 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 6};
#else
        {0, 0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 6, 0, 0, 0};
#endif
        auto expected_data = array->data();
        auto test_data = ReplaceBuffers(expected_data, 1, offset1);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);

        array = assert_array_from_json(large_binary(), R"(["01234", null, "567"])");
        const std::vector<uint8_t> offset2 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 5,
                 0, 0, 0, 0, 0, 0, 0, 5, 0, 0, 0, 0, 0, 0, 0, 8};
#else
                                                                                                                                {0, 0, 0, 0, 0, 0, 0, 0, 5, 0, 0, 0, 0, 0, 0, 0,
       5, 0, 0, 0, 0, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0, 0};
#endif
        expected_data = array->data();
        test_data = ReplaceBuffers(expected_data, 1, offset2);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);

        array = assert_array_from_json(fixed_size_binary(3), R"(["012", null, "345"])");
        expected_data = array->data();
        AssertArrayDataEqualsWithSwapEndian(expected_data, expected_data);
    }

    TEST(TestSwapEndianArrayData, StringType) {
        auto array = assert_array_from_json(utf8(), R"(["ABCD", null, "EF"])");
        const std::vector<uint8_t> offset1 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 6};
#else
        {0, 0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 6, 0, 0, 0};
#endif
        auto expected_data = array->data();
        auto test_data = ReplaceBuffers(expected_data, 1, offset1);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);

        array = assert_array_from_json(large_utf8(), R"(["ABCDE", null, "FGH"])");
        const std::vector<uint8_t> offset2 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 5,
                 0, 0, 0, 0, 0, 0, 0, 5, 0, 0, 0, 0, 0, 0, 0, 8};
#else
                                                                                                                                {0, 0, 0, 0, 0, 0, 0, 0, 5, 0, 0, 0, 0, 0, 0, 0,
       5, 0, 0, 0, 0, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0, 0};
#endif
        expected_data = array->data();
        test_data = ReplaceBuffers(expected_data, 1, offset2);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);
    }

    TEST(TestSwapEndianArrayData, ListType) {
        auto type1 = std::make_shared<ListType>(int32());
        auto array = assert_array_from_json(type1, "[[0, 1, 2, 3], null, [4, 5]]");
        const std::vector<uint8_t> offset1 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 6};
#else
        {0, 0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 6, 0, 0, 0};
#endif
        const std::vector<uint8_t> data1 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0, 4, 0, 0, 0, 5};
#else
        {0, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0, 4, 0, 0, 0, 5, 0, 0, 0};
#endif
        auto expected_data = array->data();
        auto test_data = ReplaceBuffers(expected_data, 1, offset1);
        test_data = ReplaceBuffersInChild(test_data, 0, data1);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);

        auto type2 = std::make_shared<LargeListType>(int64());
        array = assert_array_from_json(type2, "[[0, 1, 2], null, [3]]");
        const std::vector<uint8_t> offset2 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 3,
                 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0, 4};
#else
                                                                                                                                {0, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0,
       3, 0, 0, 0, 0, 0, 0, 0, 4, 0, 0, 0, 0, 0, 0, 0};
#endif
        const std::vector<uint8_t> data2 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1,
                 0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 3};
#else
                                                                                                                                {0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0,
       2, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0};
#endif
        expected_data = array->data();
        test_data = ReplaceBuffers(expected_data, 1, offset2);
        test_data = ReplaceBuffersInChild(test_data, 0, data2);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);

        auto type3 = std::make_shared<FixedSizeListType>(int32(), 2);
        array = assert_array_from_json(type3, "[[0, 1], null, [2, 3]]");
        expected_data = array->data();
        const std::vector<uint8_t> data3 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 3};
#else
        {0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0};
#endif
        test_data = ReplaceBuffersInChild(expected_data, 0, data3);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);
    }

    TEST(TestSwapEndianArrayData, DictionaryType) {
        auto type = dictionary(int32(), int16());
        auto dict = assert_array_from_json(int16(), "[4, 5, 6, 7]");
        DictionaryArray array(type, assert_array_from_json(int32(), "[0, 2, 3]"), dict);
        auto expected_data = array.data();
        const std::vector<uint8_t> data1 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 3};
#else
        {0, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0};
#endif
        const std::vector<uint8_t> data2 =
#if NEBULA_LITTLE_ENDIAN
                {0, 4, 0, 5, 0, 6, 0, 7};
#else
        {4, 0, 5, 0, 6, 0, 7, 0};
#endif
        auto test_data = ReplaceBuffers(expected_data, 1, data1);
        test_data = ReplaceBuffersInDictionary(test_data, 1, data2);
        // dictionary must be explicitly swapped
        test_data->dictionary = *::nebula::internal::SwapEndianArrayData(test_data->dictionary);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);
    }

    TEST(TestSwapEndianArrayData, StructType) {
        auto array = assert_array_from_json(STRUCT({field("a", int32()), field("b", utf8())}),
                                   R"([{"a": 4, "b": null}, {"a": null, "b": "foo"}])");
        auto expected_data = array->data();
        const std::vector<uint8_t> data1 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 4, 0, 0, 0, 0};
#else
        {4, 0, 0, 0, 0, 0, 0, 0};
#endif
        const std::vector<uint8_t> data2 =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 3};
#else
        {0, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0};
#endif
        auto test_data = ReplaceBuffersInChild(expected_data, 0, data1);
        test_data = ReplaceBuffersInChild(test_data, 1, data2);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);
    }

    TEST(TestSwapEndianArrayData, UnionType) {
        auto expected_i8 = assert_array_from_json(int8(), "[127, null, null, null, null]");
        auto expected_str = assert_array_from_json(utf8(), R"([null, "abcd", null, null, ""])");
        auto expected_i32 = assert_array_from_json(int32(), "[null, null, 1, 2, null]");
        std::vector<uint8_t> expected_types_vector;
        expected_types_vector.push_back(Type::INT8);
        expected_types_vector.insert(expected_types_vector.end(), 2, Type::STRING);
        expected_types_vector.insert(expected_types_vector.end(), 2, Type::INT32);
        std::shared_ptr<Array> expected_types;
        ArrayFromVector<Int8Type, uint8_t>(expected_types_vector, &expected_types);
        auto arr1 = SparseUnionArray::create(
                *expected_types, {expected_i8, expected_str, expected_i32}, {"i8", "str", "i32"},
                {Type::INT8, Type::STRING, Type::INT32});
        auto expected_data = (*arr1)->data();
        const std::vector<uint8_t> data1a =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 4};
#else
        {0, 0, 0, 0, 0, 0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0};
#endif
        const std::vector<uint8_t> data1b =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 0};
#else
        {0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0};
#endif
        auto test_data = ReplaceBuffersInChild(expected_data, 1, data1a);
        test_data = ReplaceBuffersInChild(test_data, 2, data1b);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);

        expected_i8 = assert_array_from_json(int8(), "[33, 10, -10]");
        expected_str = assert_array_from_json(utf8(), R"(["abc", "", "def"])");
        expected_i32 = assert_array_from_json(int32(), "[1, -259, 2]");
        auto expected_offsets = assert_array_from_json(int32(), "[0, 0, 0, 1, 1, 1, 2, 2, 2]");
        auto arr2 = DenseUnionArray::create(
                *expected_types, *expected_offsets, {expected_i8, expected_str, expected_i32},
                {"i8", "str", "i32"}, {Type::INT8, Type::STRING, Type::INT32});
        expected_data = (*arr2)->data();
        const std::vector<uint8_t> data2a =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0,
                 0, 1, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 2, 0, 0, 0, 2};
#else
                                                                                                                                {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 1, 0,
       0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 2, 0, 0, 0, 2, 0, 0, 0};
#endif
        const std::vector<uint8_t> data2b =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 3, 0, 0, 0, 6};
#else
        {0, 0, 0, 0, 3, 0, 0, 0, 3, 0, 0, 0, 6, 0, 0, 0};
#endif
        const std::vector<uint8_t> data2c =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 1, 255, 255, 254, 253, 0, 0, 0, 2};
#else
        {1, 0, 0, 0, 253, 254, 255, 255, 2, 0, 0, 0};
#endif
        test_data = ReplaceBuffers(expected_data, 2, data2a);
        test_data = ReplaceBuffersInChild(test_data, 1, data2b);
        test_data = ReplaceBuffersInChild(test_data, 2, data2c);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);
    }

    TEST(TestSwapEndianArrayData, ExtensionType) {
        auto array_int16 = assert_array_from_json(int16(), "[0, 1, 2, 3]");
        auto ext_data = array_int16->data()->copy();
        ext_data->type = std::make_shared<SmallintType>();
        auto array = make_array(ext_data);
        auto expected_data = array->data();
        const std::vector<uint8_t> data =
#if NEBULA_LITTLE_ENDIAN
                {0, 0, 0, 1, 0, 2, 0, 3};
#else
        {0, 0, 1, 0, 2, 0, 3, 0};
#endif
        auto test_data = ReplaceBuffers(expected_data, 1, data);
        AssertArrayDataEqualsWithSwapEndian(test_data, expected_data);
    }

    TEST(TestSwapEndianArrayData, MonthDayNanoInterval) {
        auto array = assert_array_from_json(month_day_nano_interval(), R"([[0, 1, 2],
                                                          [5000, 200, 3000000000]])");
        auto expected_array =
                assert_array_from_json(month_day_nano_interval(), R"([[0, 16777216, 144115188075855872],
                                                          [-2012020736, -939524096, 26688110733557760]])");

        auto swap_array = make_array(*::nebula::internal::SwapEndianArrayData(array->data()));
        EXPECT_TRUE(!swap_array->equals(array));
        ASSERT_ARRAYS_EQUAL(*swap_array, *expected_array);
        ASSERT_ARRAYS_EQUAL(
                *make_array(*::nebula::internal::SwapEndianArrayData(swap_array->data())), *array);
        ASSERT_OK(swap_array->validate_full());
    }

    DataTypeVector SwappableTypes() {
        return DataTypeVector{int8(),
                              int16(),
                              int32(),
                              int64(),
                              uint8(),
                              uint16(),
                              uint32(),
                              uint64(),
                              decimal128(19, 4),
                              decimal256(37, 8),
                              timestamp(TimeUnit::MICRO, ""),
                              time32(TimeUnit::SECOND),
                              time64(TimeUnit::NANO),
                              date32(),
                              date64(),
                              day_time_interval(),
                              month_interval(),
                              month_day_nano_interval(),
                              binary(),
                              utf8(),
                              large_binary(),
                              large_utf8(),
                              list(int16()),
                              large_list(int16()),
                              list_view(int16()),
                              large_list_view(int16()),
                              dictionary(int16(), utf8())};
    }

    TEST(TestSwapEndianArrayData, RandomData) {
        random::RandomArrayGenerator rng(42);

        for (const auto &type: SwappableTypes()) {
            NEBULA_SCOPED_TRACE("type = ", type->to_string());
            auto arr = rng.ArrayOf(*field("", type), /*size=*/31);
            ASSERT_OK_AND_ASSIGN(auto swapped_data,
                                 ::nebula::internal::SwapEndianArrayData(arr->data()));
            auto swapped = make_array(swapped_data);
            ASSERT_OK_AND_ASSIGN(auto roundtripped_data,
                                 ::nebula::internal::SwapEndianArrayData(swapped_data));
            auto roundtripped = make_array(roundtripped_data);
            ASSERT_OK(roundtripped->validate_full());

            AssertArraysEqual(*arr, *roundtripped, /*verbose=*/true);
            if (type->id() == Type::INT8 || type->id() == Type::UINT8) {
                AssertArraysEqual(*arr, *swapped, /*verbose=*/true);
            } else {
                // Random generated data is unlikely to be made of byte-palindromes
                ASSERT_FALSE(arr->equals(*swapped));
            }
        }
    }

    TEST(TestSwapEndianArrayData, InvalidLength) {
        // IPC-incoming data may be invalid, SwapEndianArrayData shouldn't crash
        // by accessing memory out of bounds.
        random::RandomArrayGenerator rng(42);

        for (const auto &type: SwappableTypes()) {
            NEBULA_SCOPED_TRACE("type = ", type->to_string());
            ASSERT_OK_AND_ASSIGN(auto arr, MakeArrayOfNull(type, 0));
            auto data = arr->data();
            // Fake length
            data->length = 123456789;
            ASSERT_OK_AND_ASSIGN(auto swapped_data, ::nebula::internal::SwapEndianArrayData(data));
            auto swapped = make_array(swapped_data);
            ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, swapped->validate());
        }
    }

    template<typename PType>
    class TestPrimitiveArray : public ::testing::Test {
    public:
        using ElementType = typename PType::T;

        void SetUp() {
            pool_ = default_memory_pool();
            GenerateInput();
        }

        void GenerateInput() {
            validity_ = std::vector<bool>{true, false, true, true, false, true};
            values_ = std::vector<ElementType>{0, 1, 1, 0, 1, 1};
        }

    protected:
        MemoryPool *pool_;
        std::vector<bool> validity_;
        std::vector<ElementType> values_;
    };

    template<>
    void TestPrimitiveArray<PDayTimeInterval>::GenerateInput() {
        validity_ = std::vector<bool>{true, false};
        values_ = std::vector<DayTimeIntervalType::DayMilliseconds>{{0, 10},
                                                                    {1, 0}};
    }

    template<>
    void TestPrimitiveArray<PMonthDayNanoInterval>::GenerateInput() {
        validity_ = std::vector<bool>{false, true};
        values_ =
                std::vector<MonthDayNanoIntervalType::MonthDayNanos>{{0, 10, 100},
                                                                     {1, 0,  10}};
    }

    TYPED_TEST_SUITE(TestPrimitiveArray, Primitives);

    TYPED_TEST(TestPrimitiveArray, IndexOperator) {
        typename TypeParam::BuilderType builder;
        ASSERT_OK(builder.Reserve(this->values_.size()));
        ASSERT_OK(builder.append_values(this->values_, this->validity_));
        ASSERT_OK_AND_ASSIGN(auto array, builder.finish());

        const auto &carray = turbo::checked_cast<typename TypeParam::ArrayType &>(*array);

        ASSERT_EQ(this->values_.size(), carray.length());
        for (int64_t i = 0; i < carray.length(); ++i) {
            auto res = carray[i];
            if (this->validity_[i]) {
                ASSERT_TRUE(res.has_value());
                ASSERT_EQ(this->values_[i], res.value());
            } else {
                ASSERT_FALSE(res.has_value());
                ASSERT_EQ(res, std::nullopt);
            }
        }
    }

}  // namespace nebula
